Что содержит foreachBatches в потоковом запросе из нескольких тем Kafka?

Question

Учитывая DataStreamReader, настроенный для подписки на несколько тем, подобных этой (см. здесь ):

// Subscribe to multiple topics
spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2,topic3")

Когда я использую foreachBatch поверх этого, что будетПакеты содержат?

• Каждый пакет будет содержать сообщения только из одной темы?
• Или пакет может содержать сообщения из разных тем?

В моемВ случае использования я хотел бы получать пакеты с сообщениями только из одной темы.Можно ли это настроить?

Answer 1

Цитирование официальной документации в Structured Streaming + Руководство по интеграции Kafka (версия брокера Kafka 0.10.0 или выше) :

// Подписаться на несколько тем

...
.option("subscribe", "topic1,topic2")

Приведенный выше код - это то, на что подписывается базовый потребитель Kafka (потокового запроса).

Когда я использую foreachBatch поверх этого, что будут содержать пакеты?

• Каждый пакет будет содержать сообщения только из одной темы?

Это правильный ответ.

Я бы хотелесть пакеты с сообщениями только из одной темы.Можно ли это настроить?

Это также описано в Руководство по интеграции структурированного потокового вещания + Kafka (версия 0.10.0 брокера Kafka или выше) :

Каждая строка в источнике имеет следующую схему:

...

topic

Другими словами, входной набор данных будет иметь столбец topicс названием темы, из которой происходит данная строка (запись).

Чтобы иметь "пакетов с сообщениями только из одной темы" вы просто filter или whereс одной темой, например

val messages: DataFrame = ...
assert(messages.isStreaming)

messages
  .writeStream
  .foreachBatch { case (df, batchId) =>
    val topic1Only = df.where($"topic" === "topic1")
    val topic2Only = df.where($"topic" === "topic2")
    ...
  }

Answer 2

Пакет будет содержать сообщения из всех тем (я бы сказал, разделы), на которые подписан ваш потребитель.