Spark结构化流式多线程/多消费者

Question

我正在使用spark结构化流媒体、合流开源Kafka集群开发spark流媒体应用程序，并在AWS EMR中运行spark job。我们至少有20个Kafka主题，以AVRO格式将数据生成单个Kafka主题，每个主题在3到4个分区之间进行了分区。我正在使用SparkReadStream阅读所有20个主题（逗号分隔的主题值）。然后从生成的数据帧中过滤每个消息行，使用正确的Avro模式应用每个消息，并将生成的数据集[T]写入S3和Cassandra。

一些我找不到答案的问题

>

火花应用程序如何在分区之间分配处理？火花是否使用不同的执行器并行读取这些主题/分区，或者我是否需要为每个分区实现任何多线程？

是否可以扩展到一个消费者组中的多个消费者以并行化？

抱歉问了这么多问题，我想这些问题都是相关的。感谢您的任何反馈或指点，我可以在那里找到文档。

>

MyConfig

val kafkaParams=  Map("kafka.bootstrap.servers" -> "topic1,topic2,topic3,topic4,topic5,
    "failOnDataLoss" -> param.fail_on_data_loss.toString,
    "subscribe" -> param.topics.toString,
    "startingOffsets" -> param.starting_offsets.toString,
    "kafka.security.protocol" -> param.kafka_security_protocol.toString,
    "kafka.ssl.truststore.location" -> param.kafka_ssl_truststore_location.toString,
    "kafka.ssl.truststore.password" -> param.kafka_ssl_truststore_password.toString
  )

ReadStream代码

val df = sparkSession.readStream
  .format("kafka")
  .options(kafkaParams)
  .load()

然后使用“主题”列将输入数据帧拆分为多个数据帧，并为每个生成的数据帧应用Avro模式。

将每个Dataset[T]写入不同的接收器，如S3、Cassandra等...

Answer 1

我可以为所有主题使用一个ReadStream吗？

假设所有主题都可以使用同一组Kafka配置，那么当然。但可能不具备容错能力。例如，failOnDataLoss将导致整个作业在单个主题失败时失败。

它会被认为是所有主题和分区的一个火花消费者吗。。。spark是否使用不同的执行器并行读取这些主题/分区？

是的。您可以将Spark执行器的数量扩展到所有主题的分区总和。

我需要为每个分区实现任何多线程吗？

Spark应该帮你解决这个问题。

是否可以扩展到一个消费者组中的多个消费者以并行化？

你应该试着设置一个组。id属性，但有多个执行者已经在创建一个消费者组。

与这个问题无关——你想做的已经是Kafka连接的目的了。将Kafka数据读入各种数据源。S3和Cassandra已经是两个现有的插件实现。