我正在将一个PySpark数据框合并到Delta表中。输出的Delta表按日期进行分区。以下查询需要30秒才能运行:
query = DeltaTable.forPath(spark, PATH_TO_THE_TABLE).alias(
"actual"
).merge(
spark_df.alias("sdf"),
"actual.DATE >= current_date() - INTERVAL 1 DAYS
AND (actual.feat1 = sdf.feat1)
AND (actual.TIME = sdf.TIME)
AND (actual.feat2 = sdf.feat2) "
,
).whenNotMatchedInsertAll()
将Delta内容复制到另一个位置后,使用NEW_PATH
替代PATH_TO_THE_TABLE
时,上述查询变得快60倍(即在同一集群上只需0.5秒)。以下是复制Delta的命令:
(spark.read.format("delta").load(PATH_TO_THE_TABLE).write.format(
"delta"
)
.mode("overwrite").partitionBy(["DATE"]).save(NEW_PATH))
如何让对第一个增量的查询像对新增量一样快?我了解增量有版本控制系统,我怀疑这是它需要花费大量时间的原因。我尝试过对增量表进行清理(将查询时间降至20秒),但离0.5秒仍有很大距离。
技术栈:
- Python 3.7;
- Pyspark 3.0.1;
- Databricks Runtime 7.3 LTS
NEW_PATH
是在同一类型的磁盘上吗?因为可能磁盘写入速度不同。 - Oliver Angelil