Geekly articles weekly

Klasifikasi sejumlah besar data pada Apache Spark menggunakan model pembelajaran mesin sewenang-wenang

Bagian 2: Solusi


Halo lagi! Hari ini saya akan melanjutkan kisah saya tentang bagaimana kita mengklasifikasikan sejumlah besar data di Apache Spark menggunakan model pembelajaran mesin sewenang-wenang. Di bagian pertama artikel, kami memeriksa pernyataan masalah itu sendiri, serta masalah utama yang muncul ketika mengatur interaksi antara cluster di mana data awal disimpan dan diproses, dan layanan klasifikasi eksternal. Pada bagian kedua, kami akan mempertimbangkan salah satu opsi untuk memecahkan masalah ini menggunakan pendekatan Reactive Streams dan implementasinya menggunakan pustaka akka-stream.


Konsep Aliran Reaktif


Untuk memecahkan masalah yang dijelaskan di bagian pertama, Anda bisa menggunakan pendekatan, yang disebut Reactive Streams . Hal ini memungkinkan Anda untuk mengontrol proses mentransfer aliran data antara tahapan pemrosesan, beroperasi pada kecepatan yang berbeda dan secara independen dari satu sama lain tanpa perlu buffering. Jika salah satu tahapan pemrosesan lebih lambat dari yang sebelumnya, perlu memberi sinyal pada tahapan yang lebih cepat tentang berapa banyak input data yang siap diproses saat ini. Interaksi ini disebut tekanan balik. Terdiri dari fakta bahwa tahap yang lebih cepat memproses elemen sebanyak yang diperlukan untuk tahap yang lebih lambat, dan tidak lebih, dan kemudian membebaskan sumber daya komputasi.

Secara umum, Aliran Reaktif adalah spesifikasi untuk menerapkan template Penerbit-Pelanggan . Spesifikasi ini menetapkan satu set empat antarmuka (Penerbit, Pelanggan, Pemroses, dan Berlangganan) dan kontrak untuk metode mereka.

Mari kita pertimbangkan antarmuka ini lebih detail:

public interface Publisher<T> { public void subscribe(Subscriber<? super T> s); } public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); } public interface Subscription { public void request(long n); public void cancel(); } public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { }

Ada dua sisi dari model Penerbit-Pelanggan: mentransmisikan dan menerima. Saat menerapkan Aliran Reaktif, kelas yang mengimplementasikan antarmuka Penerbit bertanggung jawab untuk transfer data, dan Pelanggan bertanggung jawab untuk menerima. Untuk menjalin komunikasi di antara mereka, Pelanggan harus terdaftar dengan Penerbit dengan memanggil metode berlangganannya. Menurut spesifikasi, setelah mendaftar Pelanggan, Penerbit harus memanggil metode dalam urutan berikut:

  1. di Berlangganan. Metode ini dipanggil segera setelah mendaftar Pelanggan dengan Penerbit. Sebagai parameter, objek Langganan diteruskan ke sana melalui mana Pelanggan akan meminta data dari Penerbit. Objek ini harus disimpan dan dipanggil hanya dalam konteks Pelanggan ini.
  2. Setelah Pelanggan meminta data dari Penerbit dengan memanggil metode permintaan pada objek Langganan yang sesuai, Penerbit dapat memanggil metode Pelanggan diNext, melewati elemen berikutnya.
  3. Pelanggan kemudian dapat secara berkala memanggil metode permintaan pada Langganan, tetapi Penerbit tidak dapat memanggil metode onNext lebih dari total yang diminta melalui metode permintaan.
  4. Jika aliran data terbatas, setelah melewati semua elemen melalui metode onNext, Penerbit harus memanggil metode onComplete.
  5. Jika kesalahan terjadi di Publisher dan pemrosesan lebih lanjut dari elemen tidak dimungkinkan, itu harus memanggil metode onError
  6. Setelah memanggil metode onComplete atau onError, interaksi lebih lanjut Penerbit dengan Pelanggan harus dikecualikan.

Metode panggilan dapat dianggap sebagai mengirim sinyal antara Penerbit dan Pelanggan. Pelanggan memberi sinyal kepada Penerbit berapa banyak elemen yang siap diproses, dan Penerbit, pada gilirannya, memberi sinyal kepadanya bahwa ada elemen berikutnya, atau tidak ada elemen lagi, atau beberapa kesalahan telah terjadi.

Untuk mengecualikan pengaruh lain dari Penerbit dan Pelanggan satu sama lain, panggilan ke semua metode yang menerapkan antarmuka Reactive Streams harus non-pemblokiran. Dalam hal ini, interaksi di antara mereka akan sepenuhnya tidak sinkron.

Rincian lebih lanjut tentang spesifikasi untuk antarmuka Reactive Streams dapat ditemukan di sini .

Dengan demikian, dengan menautkan iterator asli dan yang dihasilkan dengan mengonversinya masing-masing ke Publisher dan Pelanggan, kita dapat memecahkan masalah yang diidentifikasi di bagian sebelumnya artikel. Masalah buffer overflow antara tahapan diselesaikan dengan meminta sejumlah elemen oleh Pelanggan. Masalah penyelesaian yang berhasil atau tidak berhasil diselesaikan dengan mengirimkan sinyal ke Pelanggan melalui metode onComplete atau onError, masing-masing. Penerbit bertanggung jawab untuk mengirimkan sinyal-sinyal ini, yang dalam kasus kami harus mengontrol berapa banyak permintaan HTTP yang dikirim dan berapa banyak dari mereka yang menerima tanggapan. Setelah menerima respons terakhir dan memproses semua hasil yang datang di dalamnya, ia akan mengirim sinyal onComplete. Jika salah satu permintaan gagal, itu harus mengirim sinyal onError, dan berhenti mengirim elemen lebih lanjut ke Pelanggan, serta mengurangi elemen dari iterator asli.

Iterator yang dihasilkan harus diimplementasikan sebagai Pelanggan. Dalam hal ini, kita tidak dapat melakukannya tanpa buffer di mana elemen akan ditulis ketika metode onNext dipanggil dari antarmuka Pelanggan, dan dikurangi menggunakan metode hasNext dan selanjutnya dari antarmuka Iterator. Sebagai implementasi buffer, Anda dapat menggunakan antrian pemblokiran, misalnya, LinkedBlockedQueue.

Pembaca yang penuh perhatian akan segera mengajukan pertanyaan: mengapa antrian pemblokiran, karena menurut spesifikasi Reactive Streams, implementasi semua metode harus non-pemblokiran? Tapi ini baik-baik saja di sini: karena kami meminta Publisher untuk elemen yang ditentukan secara ketat, metode onNext akan dipanggil tidak lebih dari jumlah ini kali, dan antrian selalu dapat menambahkan elemen baru tanpa memblokir.

Di sisi lain, pemblokiran dapat terjadi ketika metode hasNext dipanggil jika ada antrian kosong. Namun, ini tidak apa-apa: metode hasNext bukan bagian dari kontrak antarmuka Pelanggan, itu didefinisikan dalam antarmuka Iterator, yang, seperti yang kami jelaskan sebelumnya, adalah struktur data pemblokiran. Saat memanggil metode berikutnya, kami mengurangi elemen berikutnya dari antrian, dan ketika ukurannya menjadi kurang dari ambang tertentu, kami akan perlu meminta bagian elemen berikutnya melalui panggilan ke metode permintaan.

Gambar 7. Interaksi asinkron dengan layanan eksternal menggunakan pendekatan Reactive Streams

Tentu saja, dalam hal ini kami tidak akan sepenuhnya menghilangkan pemblokiran panggilan. Ini disebabkan oleh ketidakcocokan paradigma antara aliran Reaktif, yang menganggap interaksi sepenuhnya tidak sinkron, dan iterator, yang harus memanggil trueN atau false ketika memanggil metode hasNext. Namun, tidak seperti interaksi sinkron dengan layanan eksternal, waktu henti akibat penguncian dapat dikurangi secara signifikan dengan meningkatkan beban keseluruhan inti prosesor.

Akan lebih mudah jika pengembang Apache Spark di versi yang akan datang menerapkan analog dari metode mapPartitions, yang berfungsi dengan Publisher dan Subscriber. Ini akan memungkinkan untuk interaksi yang sepenuhnya tidak sinkron, sehingga menghilangkan kemungkinan memblokir thread.

Akka-stream dan akka-http sebagai implementasi dari spesifikasi Reactive Streams


Saat ini, sudah ada lebih dari selusin implementasi spesifikasi Reactive Streams. Salah satu implementasi tersebut adalah modul akka-stream dari perpustakaan akka . Dalam dunia JVM akka telah memantapkan dirinya sebagai salah satu cara paling efektif untuk menulis sistem paralel dan terdistribusi. Ini dicapai karena fakta bahwa prinsip dasar yang ditetapkan dalam fondasinya adalah model aktor , yang memungkinkan Anda untuk menulis aplikasi yang sangat kompetitif tanpa kontrol langsung atas utas dan kumpulannya.

Banyak literatur telah ditulis tentang implementasi konsep aktor dalam akka, jadi kami tidak akan berhenti di sini ( situs resmi akka adalah sumber informasi yang sangat bagus, saya juga merekomendasikan akka dalam aksi ). Di sini kita akan melihat lebih dekat pada sisi implementasi teknologi di bawah JVM.

Secara umum, aktor tidak ada dengan sendirinya, tetapi membentuk sistem hierarkis. Untuk membuat sistem aktor, Anda perlu mengalokasikan sumber daya untuk itu, jadi langkah pertama ketika bekerja dengan akka adalah membuat turunan dari objek ActorSystem. Ketika ActorSystem dimulai, kumpulan utas terpisah dibuat, disebut dispatcher, di mana semua kode yang didefinisikan dalam aktor dieksekusi. Biasanya, utas tunggal mengeksekusi kode beberapa aktor, namun, jika perlu, Anda dapat mengonfigurasi dispatcher terpisah untuk sekelompok aktor tertentu (misalnya, untuk aktor yang berinteraksi langsung dengan API pemblokiran).

Salah satu tugas yang paling umum diselesaikan dengan menggunakan aktor adalah pemrosesan aliran data secara berurutan. Sebelumnya, untuk ini, perlu secara manual membangun rantai aktor dan memastikan bahwa tidak ada hambatan di antara mereka (misalnya, jika satu aktor memproses pesan lebih cepat dari yang berikutnya, maka ia mungkin memiliki limpahan antrian pesan masuk, yang mengarah ke kesalahan OutOfMemoryError).

Mulai dari versi 2.4, modul akka-stream ditambahkan ke akka, yang memungkinkan Anda untuk secara deklaratif mendefinisikan proses pemrosesan data, dan kemudian membuat aktor yang diperlukan untuk pelaksanaannya. Akka-stream juga menerapkan prinsip tekanan balik, yang menghilangkan kemungkinan meluapnya antrian pesan masuk untuk semua aktor yang terlibat dalam pemrosesan.

Elemen utama untuk mendefinisikan skema pemrosesan aliran data dalam akka-stream adalah Source, Flow dan Sink. Dengan menggabungkan mereka satu sama lain, kita mendapatkan Runnable Graph. Untuk memulai proses pemrosesan, materializer digunakan, yang menciptakan aktor yang bekerja sesuai dengan grafik yang ditentukan oleh kami (antarmuka Materializer dan implementasinya ActorMaterializer).

Mari kita bahas tahapan Source, Flow, dan Sink lebih detail. Sumber mendefinisikan sumber data. Akka-stream mendukung lebih dari selusin cara berbeda untuk membuat sumber, termasuk dari iterator:

 val featuresSource: Source[Array[Float], NotUsed] = Source.fromIterator { () => featuresIterator }

Sumber juga dapat diperoleh dengan mengonversi sumber yang ada:

 val newSource: Source[String, NotUsed] = source.map(item => transform(item))

Jika transformasi adalah operasi nontrivial, itu dapat direpresentasikan sebagai entitas Flow. Akka-stream mendukung banyak cara berbeda untuk membuat Flow. Cara termudah adalah membuat dari fungsi:

 val someFlow: Flow[String, Int, NotUsed] = Flow.fromFunction((x: String) => x.length)

Dengan menggabungkan Sumber dan Aliran, kami mendapatkan Sumber baru.

 val newSource: Source[Int, NotUsed] = oldSource.via(someFlow)

Sink digunakan sebagai tahap akhir dari pemrosesan data. Seperti dalam kasus Source, akka-stream menyediakan lebih dari selusin opsi Sink yang berbeda, misalnya, Sink.foreach melakukan operasi tertentu untuk setiap elemen, Sink.seq mengumpulkan semua elemen dalam koleksi, dll.

 val printSink: Sink[Any, Future[Done]] = Sink.foreach(println)

Sumber, Aliran, dan Sink masing-masing diparameterisasi oleh tipe input dan / atau elemen output. Selain itu, setiap tahap pemrosesan mungkin memiliki beberapa hasil pekerjaannya. Untuk ini, Sumber, Aliran dan Sink juga parameter dengan tipe tambahan yang menentukan hasil operasi. Tipe ini disebut tipe nilai terwujud. Jika operasi tidak menyiratkan adanya hasil tambahan dari pekerjaannya, misalnya, ketika kita mendefinisikan Flow melalui fungsi, maka tipe NotUsed digunakan sebagai nilai material.

Menggabungkan Sumber, Aliran, dan Sink yang diperlukan, kita mendapatkan RunnableGraph. Ini diparameterisasi oleh satu jenis, yang menentukan jenis nilai yang diperoleh sebagai hasil dari eksekusi grafik ini. Jika perlu, saat menggabungkan tahapan, Anda dapat menentukan hasil tahapan mana yang akan menjadi hasil dari seluruh grafik operasi. Secara default, hasil dari tahap Sumber diambil:

 val graph: RunnableGraph[NotUsed] = someSource.to(Sink.foreach(println))

Namun, jika hasil dari tahap Sink lebih penting bagi kami, maka kami harus secara eksplisit menunjukkan ini:

 val graph: RunnableGraph[Future[Done]] = someSource.toMat(Sink.foreach(println))(Keep.right)

Setelah kami menentukan grafik operasi, kami harus menjalankannya. Untuk melakukan ini, runnableGraph perlu memanggil metode run. Sebagai parameter, metode ini mengambil objek ActorMaterializer (yang juga bisa dalam lingkup implisit), yang bertanggung jawab untuk membuat aktor yang akan melakukan operasi. Biasanya, ActorMaterializer dibuat segera setelah pembuatan ActorSystem, melekat pada siklus hidupnya, dan menggunakannya untuk membuat aktor. Pertimbangkan sebuah contoh:

 //   ActorSystem,       implicit val system = ActorSystem(“system name”) // ,       ActorSystem implicit val materializer = ActorMaterializer() //    ,       Sink val graph: RunnableGraph[Future[immutable.Seq[Int]]] = Source.fromIterator(() => (1 to 10).iterator).toMat(Sink.seq)(Keep.right) //   ,    implicit scope. val result: Future[immutable.Seq[Int]] = graph.run()

Dalam kasus kombinasi sederhana, Anda dapat melakukannya tanpa membuat RunnableGraph yang terpisah, tetapi cukup sambungkan Source to Sink dan mulai dengan memanggil metode runWith pada Source. Metode ini juga mengasumsikan bahwa objek ActorMaterializer hadir dalam ruang lingkup implisit. Selain itu, dalam hal ini, nilai material yang ditentukan dalam Sink akan digunakan. Misalnya, menggunakan kode berikut, kami dapat mengonversi Sumber ke Penerbit dari spesifikasi Aliran Reaktif:

 val source: Source[Score, NotUsed] = Source.fromIterator(() => sourceIterator).map(item => transform(item)) val publisher: Publisher[Score] = source.runWith(Sink.asPublisher(false))

Jadi, sekarang kami telah menunjukkan bagaimana Anda bisa mendapatkan Reactive Streams Publisher dengan membuat Sumber dari iterator sumber dan melakukan beberapa transformasi pada elemen-elemennya. Sekarang kita dapat mengaitkannya dengan Pelanggan yang memasok data ke iterator yang dihasilkan. Masih mempertimbangkan pertanyaan terakhir: bagaimana mengatur interaksi HTTP dengan layanan eksternal.

Struktur akka termasuk modul akka-http , yang memungkinkan Anda untuk mengatur komunikasi non-blocking asinkron melalui HTTP. Selain itu, modul ini dibangun berdasarkan aliran akka, yang memungkinkan Anda untuk menambahkan interaksi HTTP sebagai langkah tambahan dalam grafik operasi pemrosesan aliran data.

Untuk terhubung ke layanan eksternal, akka-http menyediakan tiga antarmuka yang berbeda.

  1. Request-Level API - adalah opsi paling sederhana untuk kasus satu permintaan ke mesin arbitrer. Pada tingkat ini, koneksi HTTP dikelola sepenuhnya secara otomatis, dan dalam setiap permintaan, perlu mengirimkan alamat lengkap mesin yang menjadi tujuan permintaan tersebut.
  2. Host-Level API - cocok ketika kita tahu port mana di mesin mana yang akan kita akses. Dalam hal ini, akka-http mengendalikan kumpulan koneksi HTTP, dan dalam permintaan cukup untuk menentukan jalur relatif ke sumber daya yang diminta.
  3. Connection-Level API - memungkinkan Anda untuk mendapatkan kontrol penuh atas pengelolaan koneksi HTTP, yaitu membuka, menutup, dan mendistribusikan permintaan di seluruh koneksi.

Dalam kasus kami, alamat layanan klasifikasi diketahui oleh kami terlebih dahulu, oleh karena itu, perlu untuk mengatur interaksi HTTP hanya dengan mesin khusus ini. Oleh karena itu, API Tingkat Host adalah yang terbaik bagi kami. Sekarang, mari kita lihat bagaimana kumpulan koneksi HTTP dibuat saat menggunakannya:

 val httpFlow: Flow[(HttpRequest,Id), (Try[HttpResponse],Id), Http.HostConnectionPool] = Http().cachedHostConnectionPool[Id](hostAddress, portNumber)

Saat memanggil Http (). CachedHostConnectionPool [T] (hostAddress, portNumber) di ActorSystem, yang dalam lingkup implisit, sumber daya dialokasikan untuk membuat kumpulan koneksi, tetapi koneksi itu sendiri tidak dibuat. Sebagai hasil dari panggilan ini, Flow dikembalikan, yang menerima sepasang permintaan HTTP dan beberapa objek identifikasi Id sebagai input. Objek identifikasi diperlukan untuk mencocokkan permintaan dengan respons yang sesuai karena fakta bahwa panggilan HTTP di akka-http adalah operasi asinkron, dan urutan penerimaan respons tidak harus sesuai dengan urutan permintaan pengiriman. Oleh karena itu, pada Arus keluaran memberikan beberapa hasil permintaan dan objek identifikasi yang sesuai.

Secara langsung, koneksi HTTP dibuat ketika grafik (termasuk Flow ini) diluncurkan (terwujud). Akka-http diimplementasikan sedemikian rupa sehingga tidak peduli berapa kali grafik yang mengandung httpFlow telah terwujud, dalam ActorSystem yang sama akan selalu ada satu kumpulan koneksi HTTP yang akan digunakan oleh semua materialisasi. Ini memungkinkan Anda untuk mengontrol penggunaan sumber daya jaringan dengan lebih baik dan menghindari kelebihannya.

Dengan demikian, siklus hidup kumpulan koneksi HTTP terkait dengan ActorSystem. Seperti yang telah disebutkan, siklus hidup thread pool juga melekat padanya, di mana operasi yang didefinisikan dalam aktor dilakukan (atau dalam kasus kami, didefinisikan sebagai tahapan akka-stream dan akka-http). Oleh karena itu, untuk mencapai efisiensi maksimum, kita harus menggunakan kembali satu instance ActorSystem dalam proses JVM yang sama.

Menyatukan semua ini: contoh penerapan interaksi dengan layanan klasifikasi


Jadi, sekarang kita dapat beralih ke proses mengklasifikasikan volume besar data terdistribusi di Apache Spark menggunakan interaksi asinkron dengan layanan eksternal. Skema umum dari interaksi ini telah ditunjukkan pada Gambar 7.

Misalkan kita memiliki beberapa Dataset [Fitur] awal yang didefinisikan. Menerapkan operasi mapPartitions untuk itu, kita harus mendapatkan Dataset, di mana setiap id dari set sumber dicap dengan nilai tertentu yang diperoleh sebagai hasil klasifikasi (Dataset [Skor]). Untuk mengatur pemrosesan tidak sinkron pada pelaksana, kita harus membungkus sumber dan menghasilkan iterator di Publisher dan Pelanggan, masing-masing, dari spesifikasi aliran Reaktif dan menghubungkannya bersama.

 case class Features(id: String, vector: Array[Float]) case class Score(id: String, score: Float) //(1) val batchesRequestCount = config.getInt(“scoreService. batchesRequestCount”)//(2) //... val scoreDs: Dataset[Score] = featuresDs.mapPartitions { fi: Iterator[Features] => val publisher: Publisher[Iterable[Score]] = createPublisher(fi) //(3) val iteratorSubscriber: Iterator[Score] = new IteratorSubscriber(batchesRequestCount) //(4) publisher.subscribe(batchesRequestCount) //(5) iteratorSubscriber //(6) }

Dalam implementasi ini, diperhitungkan bahwa layanan klasifikasi untuk satu panggilan dapat memproses sekelompok vektor fitur segera, oleh karena itu, hasil klasifikasi setelah panggilan itu juga akan segera tersedia untuk seluruh grup. Karena itu, sebagai tipe parameter untuk Penerbit, kami tidak hanya Skor, seperti yang Anda harapkan, tetapi [Skor] terteratur. Jadi, kami mengirim hasil klasifikasi untuk grup ini ke iterator yang dihasilkan (yang juga merupakan Pelanggan) dengan satu panggilan ke metode onNext. Ini jauh lebih efisien daripada memanggil onNext untuk setiap elemen. Sekarang kita akan menganalisis kode ini lebih terinci.

  1. Kami menentukan struktur data input dan output. Sebagai input, kita akan memiliki banyak pengidentifikasi id dengan vektor fitur, dan sebagai output, kita akan memiliki banyak pengidentifikasi dengan nilai numerik yang diperoleh sebagai hasil klasifikasi.
  2. Kami menentukan jumlah grup yang akan diminta oleh Pelanggan dari Penerbit sekaligus. Karena diasumsikan bahwa nilai-nilai ini akan terletak di buffer dan menunggu sampai mereka dibaca dari iterator yang dihasilkan, nilai ini tergantung pada jumlah memori yang dialokasikan untuk pelaksana.
  3. Buat Penerbit dari sumber iterator. Dia akan bertanggung jawab untuk berinteraksi dengan layanan klasifikasi. Fungsi createPublisher dibahas di bawah ini.
  4. Buat Pelanggan, yang akan menjadi iterator yang dihasilkan. Kode kelas IteratorSubscriber juga diberikan di bawah ini.
  5. Mendaftarkan Pelanggan ke Penerbit.
  6. Kembalikan IteratorSubscriber sebagai hasil dari operasi mapPartitions.

Sekarang pertimbangkan penerapan fungsi createPublisher.

 type Ids = Seq[String] //(1) val batchSize = config.getInt("scoreService.batchSize") val parallelismLevel = config.getInt("scoreService.parallelismLevel") //(2) //... def createPublisher(fi: Iterator[Features]): Publisher[Iterable[Score]] = { import ActorSystemHolder._ //(3) Source .fromIterator(() => fi) //(4) .grouped(batchSize) //(5) .map { groupedFeatures: Seq[Features] => val request: (HttpRequest, Ids) = createHttpRequest(groupedFeatures) //(6) logger.debug(s"Sending request for the first id: ${request._2(0)}") request } .via(httpFlow) //(7) .flatMapMerge(parallelismLevel, { //(8) case (Success(response), ids) if response.status.isSuccess() => logger.debug(s"Processing successful result for the first id: ${ids(0)}") val resultSource: Source[Iterable[Score], _] = response.entity.dataBytes.reduce(_ ++ _).map { responseBytes => processSuccessfulResponse(responseBytes, ids) } //(9) resultScore case (Success(response), ids) => logger.warn( s"Failed result for the first id: ${ids(0)}, HTTP status: ${response.status}" ) response.discardEntityBytes() Source.failed( new IOException(s"Non-successful HTTP status: ${response.status}") ) //(10) case (Failure(ex), ids) => logger.warn(s"Failed result: an exception has occured", ex) Source.failed(ex) //(11) }) .runWith(Sink.asPublisher(false)) //(12) } def createHttpRequest(featuresSeq: Seq[Features]): (HttpRequest, ProfileIds) = { val requestBytes: Array[Byte] = featuresToMatrixBytes(featuresSeq) val ids: ProfileIds = extractIds(featuresSeq) val httpRequest = HttpRequest( method = HttpMethods.PUT, uri = "/score", entity = requestBytes ) httpRequest -> ids }

  1. - , . httpFlow, .
  2. : , (batchSize) (parallelismLevel).
  3. implicit scope ActorSystem, ActorMaterializer httpFlow. Spark-. ActorSystemHolder .
  4. akka-streams . Source[Features] .
  5. batchSize .
  6. HttpRequest . HttpRequest createHttpRequest. createPublisher. feature-, , ( predict). , HTTP-. , HTTP-, HTTP-, URI .
  7. httpFlow.
  8. , . flatMapMerge, akka-http Source[ByteString], , . . parallelismLevel , ( ). HTTP-: , , , .
  9. : . akka ByteString. , ByteString O(1), ByteString . , , . , .
  10. HTTP- , Stream . , discardEntityBytes , , .
  11. . akka-http , .
  12. , Publisher, . , . false Sink.asPublisher , Publisher Subscriber-.

, akka ActorSystem, . , Spark , . Spark JVM , , , ActorSystem ActorMatrializer httpFlow.

 object ActorSystemHolder { implicit lazy val actorSystem: ActorSystem = { //(1) val actorSystemName = s"score-service-client" logger.debug(s"Creating actor system $actorSystemName") val as = ActorSystem(actorSystemName) //(2) logger.debug("Adding shutdown hook for the actor system") scala.sys.addShutdownHook { //(3) logger.debug(s"Terminating actor system $actorSystemName") Await.result(as.terminate(), 30.seconds) //to Mars :) logger.debug(s"The actor system $actorSystemName has been terminated") } as } implicit lazy val materializer: ActorMaterializer = { //(4) logger.debug(s"Creating actor materializer for actor system ${actorSystem.name}") ActorMaterializer() } lazy val httpFlow: Flow[ (HttpRequest,ProfileIds), (Try[HttpResponse], ProfileIds), Http.HostConnectionPool] = { //(5) val httpFlowSettings = ConnectionPoolSettings(actorSystem) logger.debug(s"Creating http flow with settings $httpFlowSettings") Http().cachedHostConnectionPool[ProfileIds]( config.getString("scoreService.host"), config.getInt("scoreService.int"), settings = httpFlowSettings ) } }

  1. , , , .
  2. ActorSystem .
  3. , , ActorSystem, terminate, , , , . , JVM-.
  4. ActorMaterializer, akka-streams, ActorSystem.
  5. , httpFlow . , HTTP- ActorSystem.

Subscriber- HTTP-.

 sealed trait QueueItem[+T] case class Item[+T](item: T) extends QueueItem[T] case object Done extends QueueItem[Nothing] case class Failure(cause: Throwable) extends QueueItem[Nothing] //(1) class StreamErrorCompletionException(cause: Throwable) extends Exception(cause) //(2) class IteratorSubscriber[T](requestSize: Int) extends Subscriber[Iterable[T]] with Iterator[T] { //(3) private val buffer: BlockingQueue[QueueItem[Iterable[T]]] = new LinkedBlockingQueue[QueueItem[Iterable[T]]]() //(4) private var expecting: Int = 0 //(5) private val subscriptionPromise: Promise[Subscription] = Promise() private lazy val subscription: Subscription = Await.result(subscriptionPromise.future, 5.minutes) //(6) private var currentIterator: Iterator[T] = Iterator.empty //(7) private var isDone = false //(8) override def onSubscribe(s: Subscription): Unit = { subscriptionPromise.success(s) //(9) logger.trace("The iterator has been subscribed") } override def onNext(t: Iterable[T]): Unit = { logger.trace("Putting a next batch to the buffer") buffer.put(Item(t)) //(10) } override def onComplete(): Unit = { logger.debug("The stream has been succesfully completed") buffer.put(Done) //(11) } override def onError(t: Throwable): Unit = { logger.warn("The stream has been completed with error", t) buffer.put(Failure(t)) //(12) } override def hasNext: Boolean = { logger.trace("Asking hasNext") if (currentIterator.hasNext) { //(13) true } else if (isDone) { //(14) false } else { if (expecting < requestSize) { requestNextBatches() //(15) } buffer.take() match { //(16) case Item(batch) => currentIterator = batch.iterator expecting -= 1 this.hasNext //(17) case Done => isDone = true false //(18) case Failure(exception) => throw new StreamErrorCompletionException(exception) //(19) } } } override def next(): T = { val out = currentIterator.next() logger.trace("The next element is {}", out) out //(20) } private def requestNextBatches(): Unit = { logger.debug(s"Requesting {} batches", requestSize) subscription.request(requestSize) expecting += requestSize //(21) } }

IteratorSubscriber Producer-Consumer. , Subscriber, Producer-, , Iterator, – Consumer-. , . Iterator Apache Spark, Subscriber – , ActorSystem.

IteratorSubscriber .

  1. . , Done, , Throwable, .
  2. , hasNext .
  3. , , Publisher-.
  4. , . LinkedBlockingQueue, . , .
  5. , . , , Publisher-. , , Publisher- . hasNext next ( requestNextBatches hasNext), , .
  6. subscriptionPromise subscription Subscription, Publisher onSubscribe. , Reactive Streams Subscriber- Publisher- , , hasNext , onSubscribe. , subscription, Publisher-. lazy subscription, Promise.
  7. . hasNext next, , .
  8. , , hasNext false . hasNext, .
  9. onSubscribe Publisher- Subscription Promise, subscription.
  10. onNext Publisher-, . .
  11. Publisher onComplete, Done.
  12. Publisher onError. .
  13. hasNext , . , true, . , .
  14. , false.
  15. , , requestSize, Publisher. , , , Publisher- , HTTP- .
  16. . , , , . , , ( , , subscription), , , , .
  17. , currentIterator. , . , hasNext , ( , ), .
  18. , false hasNext. , isDone, , . - , hasNext , false. , hasNext , false , . , .
  19. , , , .
  20. next . , hasNext, next .
  21. Publisher- , , subscription, Publisher-. requestSize. .

, , , :

8. .

:


, , . , HTTP , . .

– . , , Hadoop , . , , - . , , hdfs, , , , .

, . , akka-http , . , -, - Apache Spark , , , -.

, , . , , http-, , .

, . , , . , . , .

, . , , Hadoop , , .

, , Hadoop- , , .

, , CleverDATA . . , , , , , . , .

, .

Source: https://habr.com/ru/post/id413141/

More articles:


All Articles