全部推送完数据后批量消费Kafka答案

【问题标题】：Consume Kafka in batches when all data has been pushed全部推送完数据后批量消费Kafka
【发布时间】：2018-02-06 21:41:12
【问题描述】：

我是 Kafka 新手，我想做以下事情：

我有一堆服务器每 10 分钟将一些数据推送到 Kafka。
我有一个 Spark 应用程序，它需要所有服务器推送的最新数据。

例如：我有 2 个推送服务器，分别是 'a' 和 'b'。我需要 spark 应用程序在数据框中接收值 'a' 和 'b' 以便它们可以一起处理。 10 分钟后，两台服务器推送'c' 和'd'。 spark 应用应该同时接收到值'c' 和'd'，等等。

我的 Spark 应用程序需要推送所有最新的数据，因此我认为流式方法是不正确的，可能应该采用批处理方法（或者它的名称可能不同）。

我的 Spark 应用需要 DataFrame。

【问题讨论】：

没有任何保证。您需要在您的应用中自行组装数据框。
假设我可以做到，我仍然需要以某种格式一起接收所有数据，这是我真正的问题。
您可以配置您的 kafka 消费者以查找最新消息。 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG，“最新”people.apache.org/~nehanarkhede/kafka-0.9-producer-javadoc/doc/…

标签： pyspark apache-kafka kafka-consumer-api kafka-producer-api

【解决方案1】：

您的问题听起来不像是常见的 Kafka 用例。但是，如果必须使用 Kafka，您可以使用 Kafka 主题对数据进行分组。通过创建主题A_B 和C_D，您可以确保值'a' 和'b' 将一起使用并与'c' 和'd' 值分开。然后，您的 Spark 应用程序必须验证它是否从 A_B 和 C_D 获得了所有需要的数据并继续执行。如果您的 Spark 应用程序能够缓冲所有数据并确定何时消耗了所有需要的消息，则此设计将起作用。

【讨论】：

【解决方案2】：

首次加载应用时，需要保存 Kafka 主题的当前偏移量。然后，每 10 分钟从 Kafka 加载数据并执行逻辑。所以基本上可以运行如下例程：

存储当前偏移量
等待 10 分钟
初始化 DataFrame 并从 Kafka 加载数据（在您的情况下，它将首次加载“a”和“b”）
执行你的逻辑

当您加载 DataFrame 时，您可以指定偏移量，从而确保您获得过去 10 分钟内收到的数据。

例如，您可以执行以下操作来初始化 DataFrame：

df = spark.read.format("kafka") \
     .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
     .option("subscribe", "your_topic") \
     .option("startingOffsets", """{"your_topic":{"0":saved_offset}}""") \
     .option("endingOffsets", """{"your_topic":{"0":-1}}""") \
     .load()

或者，如果您想使用流式应用程序，想法几乎相同。但是，在这种情况下，您从当前的 Kafka 位置开始（或存储一次起始偏移量并从它开始），然后您运行的每个附加查询将在前一个查询停止的地方继续。因此，在这种情况下，您首先初始化 DataFrame，然后运行以下例程：

等待 10 分钟
查询 DataFrame 并执行您的逻辑

在这种情况下初始化DataFrame如下：

df = spark.readStream.format("kafka") \
     .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
     .option("subscribe", "your_topic") \
     .load()

您可以在此处获取更多信息： https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html

【讨论】：