Как сохранить данные Кафки в другое место на основе значения столбца в потоковой структурированной искре?

Question

У меня есть сценарий использования, в котором я использую данные из Кафки с использованием потоковой структурированной искры.У меня есть несколько тем для подписки, и на основе имени topic датафрейм должен быть выгружен в определенное место (другое место для разных тем).Я видел, может ли это быть решено с помощью какой-либо функции разделения / фильтрации в кадре данных spark, но не смог найти ни одной.

На данный момент у меня есть подписка только на одну тему, и я использую свой собственный письменный метод для выгрузки данных в место в формате паркета.Вот код, который я сейчас использую:

def save_as_parquet(cast_dataframe: DataFrame,output_path: 
      String,checkpointLocation: String): Unit = {
  val query = cast_dataframe.writeStream
              .format("parquet")
              .option("failOnDataLoss",true)
              .option("path",output_path)
              .option("checkpointLocation",checkpointLocation)
              .start()
              .awaitTermination()
 }

Когда я буду подписан на разные темы, этот cast_dataframe также будет иметь значения из разных тем.Я хочу сбросить данные из темы только в то место, где ей назначено местоположение.Как это можно сделать?

Answer 1

Как объяснено в официальной документации Dataset, которая должна быть написана, может содержать необязательный столбец topic, который можно использовать для маршрутизации сообщений:

* Столбец темы необходим, если не указан параметр конфигурации «тема».

Столбец значения является единственной обязательной опцией. Если ключевой столбец не указан, то ключевой столбец с нулевым значением будет добавлен автоматически (см. Семантику Kafka о том, как обрабатываются значения ключа с нулевым значением). Если столбец темы существует, то его значение используется в качестве темы при записи данной строки в Kafka, если параметр конфигурации «темы» не установлен, т.е. параметр конфигурации «темы» переопределяет столбец темы.