Java 中的 Spark Streaming：使用 JavaInputDStream 使用一个消费者从两个 Kafka 主题中读取答案

【问题标题】：Spark Streaming in Java: Reading from two Kafka Topics using One Consumer using JavaInputDStreamJava 中的 Spark Streaming：使用 JavaInputDStream 使用一个消费者从两个 Kafka 主题中读取
【发布时间】：2020-06-30 04:20:36
【问题描述】：

我有一个 spark 应用程序，它需要使用一个使用 Spark Java 的消费者从两个不同的主题中读取。两个主题的 kafka 消息键和值架构相同。

以下是工作流程：

1. Read messages from both the topics, same groupID, using JavaInputDStream<ConsumerRecord<String, String>> and iterate using foreachRDD
2. Inside the loop, Read offsets, filter messages based on the message key and create JavaRDD<String>
3. Iterate on JavaRDD<String> using mapPartitions
4. Inside mapPartitions loop, iterate over them using forEachRemaining.
5. Perform data enrichment, transformation, etc on the rows inside forEachRemaining loop.
6. commit

我想了解以下问题。请提供您的答案或分享任何可以帮助我找到答案的文档。

1. How the messages are received/consumed from two topics(one common group id, same schema both key/value) in one consumer.
Let say the consumer reads data every second. Producer1 produces 50 messages to Topic1 and Producer 2 produces 1000 messages to Topic2.
2. Is it going to read all msgs(1000+50) in one batch and process together in the workflow, OR is it going to read 50 msgs first, process them and then read 1000 msgs and process them.
3. What parameter should i use to control the number of messages being read in one batch per second.
4. Will same group id create any issue while consuming.

【问题讨论】：

标签： apache-spark apache-kafka spark-streaming kafka-consumer-api

【解决方案1】：

Spark Streaming 中的官方文档已经解释了如何为每个组 id 消费多个主题。 https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

Collection<String> topics = Arrays.asList("topicA", "topicB");

JavaInputDStream<ConsumerRecord<String, String>> stream =
  KafkaUtils.createDirectStream(
    streamingContext,
    LocationStrategies.PreferConsistent(),
    ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
  );

一个组 id 并遵循两个主题的相同架构。
对此不确定，但据我了解，它会根据批处理大小消耗所有消息。
“spark.streaming.backpressure.enabled”将此设置为 true，“spark.streaming.kafka.maxRatePerPartition”将其设置为数值，基于此 spark 限制每批从 kafka 消耗的消息数量。还相应地设置批处理持续时间。 https://spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/streaming/api/java/JavaStreamingContext.html
这完全取决于您的应用程序使用情况。

【讨论】：

【解决方案2】：

1.如何从一个消费者中的两个主题（一个公共组 id，相同的模式和键/值）接收/使用消息。假设消费者每秒读取一次数据。 Producer1 向 Topic1 生成 50 条消息，Producer 2 向 Topic2 生成 1000 条消息。
任何 Kafka 消费者都可以提及主题列表，因此对此没有任何限制。
因此，如果您有一个消费者，它将负责 Topic1 和 Topic2 的所有分区。
2。是批量读取所有消息（1000+50）并在工作流程中一起处理，还是先读取 50 条消息，处理它们，然后读取 1000 条消息并处理它们。
3。我应该使用什么参数来控制每秒一批读取的消息数。
回答 2,3 个问题：
它将一起接收所有消息 (1050) 甚至更多，具体取决于您的配置。
为了让消费者分批接收1050或更大，将max.poll.records（默认500）提高到1050（或更多）；其他配置可能是一个瓶颈，但对于默认配置，您应该可以接受其余配置。
4。使用相同的组 ID 是否会产生任何问题。
如果您创建多个消费者，相同的 group-id 将影响您，使消费者在主题之间拆分他们负责的分区。
此外，如果您的消费者因某种原因死亡或停止，您必须使用相同的 group-id 恢复它，这样消费者“记住”最后消耗的偏移量并保持它停止的点。

如果您对您的消费者有任何问题，我建议您阅读此article 中的更多信息，这是 Kafka：权威指南的第 4 章，深入解释了消费者并应回答更多问题。
如果您想探索配置选项，documentation 总是有帮助的。

【讨论】：