如何基于分区和聚簇列从Apache Spark覆盖BigQuery指定时段数据？

要实现仅覆盖指定batch_date和batch_hour的数据，你需要避免直接使用SaveMode.Overwrite（它会覆盖整个batch_date分区），可以通过以下两种方式实现：

方法一：先删除目标数据，再追加写入 #

1. 删除指定条件的旧数据：通过Spark SQL连接BigQuery，执行DELETE语句清理目标日期和小时的数据

// 替换为你的项目、数据集和表名
spark.sql("DELETE FROM `your-project.your-dataset.your-table` WHERE batch_date = '2023-03-03' AND batch_hour = 8")

2. 追加写入新数据：使用SaveMode.Append将新的DataFrame写入目标表，确保只添加指定批次的数据

df.write.mode(SaveMode.Append).format("bigquery")
  .option("partitionField", "batch_date")
  .option("clusterFields", "batch_hour")
  .option("temporaryGcsBucket", "temp_bucket")
  // 其他表配置选项
  .save("your-project.your-dataset.your-table")

方法二：使用MERGE实现UPSERT（更安全） #

这种方式可以避免删除后写入失败导致的数据丢失，通过合并操作替换匹配的数据、插入新数据：

1. 将新数据写入临时表

df.write.mode(SaveMode.Overwrite).format("bigquery")
  .option("temporaryGcsBucket", "temp_bucket")
  .save("your-project.your-dataset.temp-table")

2. 执行MERGE语句

spark.sql("""
  MERGE INTO `your-project.your-dataset.your-table` target
  USING `your-project.your-dataset.temp-table` source
  ON target.batch_date = source.batch_date AND target.batch_hour = source.batch_hour
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
""")

关键说明 #

• 直接使用SaveMode.Overwrite写入整个表时，BigQuery会将整个batch_date分区替换，这就是你之前覆盖所有小时数据的原因。
• 两种方法都利用了BigQuery对聚簇列的支持，针对batch_hour的过滤和更新操作效率较高。

内容的提问来源于stack exchange，提问作者A B