scala - Akka-Stream を使用して Reactive Kafka で大きなメッセージを「チャンクして再組み立て」する方法

Question

Kafka を使用して大きなファイルを送信する場合、複数のパーティションに分散してから Akka-Stream を使用して再構築することは可能ですか? このプレゼンテーションで説明されているように:

http://www.slideshare.net/JiangjieQin/handle-large-messages-in-apache-kafka-58692297

Answer 1

「チャンキング」側、つまりプロデューサーは、リアクティブ kafkaのようなものを使用して簡単に記述できます。

case class LargeMessage(bytes : Seq[Byte], topic : String)

def messageToKafka(message : LargeMessage, maxMessageSize : Int) = 
  Source.fromIterator(() => message.bytes.toIterator)
        .via(Flow[Byte].grouped(maxMessageSize))
        .via(Flow[Seq[Byte]].map(seq => new ProducerRecord(message.topic, seq)))
        .runWith(Producer.plainSink(producerSettings)

「再組み立て」、つまり消費者は、ドキュメントと同様の方法で実装できます。

   val messageFut : Future[LargeMessage] = 
     for {
       bytes <- Consumer.map(_._1).runWith(Sink.seq[Byte])
     } yield LargeMessage(bytes, topic)