#apache-spark #spark-structured-streaming #spark-streaming-kafka
#apache-spark #spark-structured-streaming #spark-streaming-kafka
Вопрос:
Я переношу потоковое задание (раздел Kafka -> Файлы AWS S3 Parquet) из Kafka Connect в Spark Structured Streaming Job.
Я разделяю свои данные по годам / месяцам / дням.
Код очень прост:
df.withColumn("year", functions.date_format(col("createdAt"), "yyyy"))
.withColumn("month", functions.date_format(col("createdAt"), "MM"))
.withColumn("day", functions.date_format(col("createdAt"), "dd"))
.writeStream()
.trigger(processingTime='15 seconds')
.outputMode(OutputMode.Append())
.format("parquet")
.option("checkpointLocation", "/some/checkpoint/directory/")
.option("path", "/some/directory/")
.option("truncate", "false")
.partitionBy("year", "month", "day")
.start()
.awaitTermination();
Выходные файлы находятся в следующем каталоге (как и ожидалось):
/s3-bucket/some/directory/year=2021/month=01/day=02/
Вопрос:
Есть ли способ настроить имя выходного каталога? Мне нужно, чтобы это было
/s3-bucket/some/directory/2021/01/02/
По соображениям обратной совместимости.
Комментарии:
1. Kafka Connect может (и, вероятно, должен) записывать в
year=2021/month=01/day=02
формате, так какую выгоду добавляет Spark?
Ответ №1:
Нет, нет способа настроить имена выходных каталогов в формате, который вы упомянули в своем приложении Spark Structured Streaming.
Разделы основаны на значениях определенных столбцов, и без их имен столбцов в пути к каталогу было бы неоднозначно, к какому столбцу принадлежит их значение. Вам нужно написать отдельное приложение, которое преобразует эти каталоги в нужный формат.