Блог

Я переношу потоковое задание (раздел 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/
 

По соображениям обратной совместимости.

Нет, нет способа настроить имена выходных каталогов в формате, который вы упомянули в своем приложении Spark Structured Streaming.

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