如何使用接收记录的年、月和日将流写入 S3？答案

【问题标题】：How to write stream to S3 with year, month and day of the day when records were received?如何使用接收记录的年、月和日将流写入 S3？
【发布时间】：2020-11-15 22:35:10
【问题描述】：

我有一个从 Kafka 主题读取一些数据的简单流：

 val ds = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1")
      .option("subscribe", "topic1")
      .option("startingOffsets", "earliest")
      .load()

val df = ds.selectExpr("cast (value as string) as json")
      .select(from_json($"json", schema).as("data"))
      .select("data.*")

我想根据收到的日期将此数据存储在 S3 中，例如：

s3_bucket/year/month/day/data.json

当我想写数据时：

df.writeStream
  .format("json")
  .outputMode("append")
  .option("path", s3_path)
  .start()

但如果我这样做，我只能指定一个路径。有没有办法根据日期动态更改 s3 路径？

【问题讨论】：

如果要按时间分区，请考虑如何查询。通常的年、月、日路径嵌套的问题在于它使某些类型的查询非常困难。通常最好沿单个轴进行分区，例如 s3_bucket/table_name/ts=yyyymmddhhmm/*.json，这允许快速范围查询。见spark-summit.org/east-2017/events/…

标签： scala apache-spark spark-structured-streaming

【解决方案1】：

使用partitionBy 子句：

import org.apache.spark.sql.functions._

df.select(
    dayofmonth(current_date()) as "day",
    month(current_date()) as "month",
    year(current_date()) as "year",
    $"*")
  .writeStream
  .partitionBy("year", "month", "day")
  ... // all other options

【讨论】：