将案例类列表连接到kafka制作人？

snvhrwxg 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(396)

我有下面的案例课：

case class Alpakka(id:Int,name:String,animal_type:String)

我正在尝试使用以下代码将这些案例类的列表连接到kafka中的生产者：

def connectEntriesToProducer(seq: Seq[Alpakka]) = {
    val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
      .withBootstrapServers("localhost:9092")
seq.map(alpakka => new ProducerRecord[String, String]("alpakkas", alpakka.asJson.noSpaces))   
      .runWith(Producer.plainSink(producerSettings))
  }

我使用circe将case类转换为json。但是，我不断收到一个编译器错误：

Error:(87, 34) type mismatch;
 found   : akka.stream.scaladsl.Sink[org.apache.kafka.clients.producer.ProducerRecord[String,String],scala.concurrent.Future[akka.Done]]
 required: org.apache.kafka.clients.producer.ProducerRecord[String,String] => ?
      .runWith(Producer.plainSink(producerSettings))

我不知道发生了什么事！

scala akka apache-kafka alpakka

来源：https://stackoverflow.com/questions/55480526/connecting-list-of-case-classes-to-kafka-producer

1条答案

按热度按时间

zwghvu4y1#

你想建立一个 Graph 从 Seq 而不是 Source .
你的方法 connectEntriesToProducer 应该看起来像

def connectEntriesToProducer(seq: Source[Alpakka]) = {

注意， Source 而不是 Seq .
或者，您可以从 Seq ，但你必须使用 immutable.Seq 自 Source.apply 只需要一个不变的iterable。

def connectEntriesToProducer(seq: scala.collection.immutable.Seq[Alpakka]) = {
val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
  .withBootstrapServers("localhost:9092")
Source(seq).
  map(alpakka => new ProducerRecord[String, String]("alpakkas", alpakka.asJson.noSpaces))
  .runWith(Producer.plainSink(producerSettings))
}

赞(0）回复(0）举报 2021-06-07

我来回答

将案例类列表连接到kafka制作人？

1条答案

相关问题

热门标签

最新问答