附加到 parquet 文件，由具有重叠时间戳的数据分区

Question

我有包含时间戳列的数据框。我将其转换为日期，按日期进行分区，并每天将其附加到不断增长的镶木地板文件中。

如果我附加一个时间戳从 2021-04-19 01:00:01 到 2021-04-19 13:00:00 的数据集，它会将其写入分区 DATE=2021-04-19 中的镶木地板。

如果在当天晚些时候我将另一个时间戳从2021-04-19 15:00:00 附加到2021-04-19 20:00:00 的数据集，它会覆盖之前从凌晨 1 点到下午 1 点有数据的分区吗？还是它实际上会附加到它上面？

我使用语法：

df.write.mode('append').partitionBy("DATE").parquet("s3://path")

Answer 1

来自Spark Documentation中的保存模式：

追加：将 DataFrame 保存到数据源时，如果数据/表已经 存在，DataFrame 的内容应附加到 现有数据。

因此，它符合您的预期。这是一个检查行为的玩具示例：

data_batch_1 = [("2021-04-19", "2021-04-19 01:00:01", 1.1), 
                ("2021-04-19", "2021-04-19 13:00:00", 1.2)]

data_batch_2 = [("2021-04-19", "2021-04-19 15:00:00", 2.1), 
                ("2021-04-19", "2021-04-19 20:00:00", 2.2)]

col_names = ["DATE", "ts", "sensor1"]

df_batch_1 = spark.createDataFrame(data_batch_1, col_names)
df_batch_2 = spark.createDataFrame(data_batch_2, col_names)

s3_path = "/tmp/67163237/"

保存批次 1

df_batch_1.write.mode("append").partitionBy("DATE").parquet(s3_path)
spark.read.parquet(s3_path).show()

+-------------------+-------+----------+
|                 ts|sensor1|      DATE|
+-------------------+-------+----------+
|2021-04-19 01:00:01|    1.1|2021-04-19|
|2021-04-19 13:00:00|    1.2|2021-04-19|
+-------------------+-------+----------+

保存批次 2

df_batch_2.write.mode("append").partitionBy("DATE").parquet(s3_path)
spark.read.parquet(s3_path).show()

+-------------------+-------+----------+
|                 ts|sensor1|      DATE|
+-------------------+-------+----------+
|2021-04-19 15:00:00|    2.1|2021-04-19|
|2021-04-19 01:00:01|    1.1|2021-04-19|
|2021-04-19 20:00:00|    2.2|2021-04-19|
|2021-04-19 13:00:00|    1.2|2021-04-19|
+-------------------+-------+----------+

Answer 2

按照 mck 的出色建议（您在尝试之前不会知道），我做到了，而且正如担心的那样，它基本上用新数据覆盖了整个分区。

我想了想，我决定总是重新流式传输前一天并覆盖分区。在我的情况下它对我有用，因为我确实可以访问 5 天的缓冲区数据，我可以重新提取这些数据。但这种解决方案不适用于那些拥有仅保留几个小时左右的临时数据的人。