apache-spark - Foundry 中的 Spark 可以使用分区修剪吗

Question

我们有一个数据集，它在我们的 Foundry 实例上作为增量构建运行。该数据集是一个大型时间序列数据集(565 亿行，10 列，965GB)，时间戳以 1 小时为单位。数据集每天增长约 10GB。

为了优化数据集以供分析，我们根据两个属性“measure_date”和“measuring_time”对数据集进行了重新分区。

这反射(reflect)了访问模式——数据集通常由“measure_date”访问。我们通过“measuring_time”对其进行子分区以减小生成的 Parquet 文件的大小，而且按时间过滤也是一种常见的访问模式。

创建分区的代码如下:

if ctx.is_incremental:
    return df.repartition(24, "measure_date", "measuring_time")
else:
    return df.repartition(2200, "measure_date", "measuring_time")

使用哈希分区会创建不平衡的文件大小，但这是另一篇文章的主题。

我现在正试图找出如何让 Spark on Foundry 利用过滤条件中的分区。据我所知，情况并非如此。

我创建了一个代码工作簿并对遥测数据运行了以下查询，将结果保存到另一个数据集。

SELECT * 
FROM telemetry_data
where measure_date = '2022-06-05'

构建的物理查询计划似乎表明，Spark 没有使用任何分区，计划中的 PartitionFilters 为空。

Batched: true, BucketedScan: false, DataFilters: [isnotnull(measure_date#170), (measure_date#170 = 19148)], 
            Format: Parquet, Location: InMemoryFileIndex[sparkfoundry://prodapp06.palantir:8101/datasets/ri.foundry.main.dataset.xxx..., 
            PartitionFilters: [],
            PushedFilters: [IsNotNull(measure_date), EqualTo(measure_date,2022-06-05)],
            ReadSchema: struct<xxx,measure_date:date,measuring_time_cet:timestamp,fxxx, ScanMode: RegularMode

如何让 Spark on Foundry 使用分区修剪？

Answer 1

我相信你需要使用transforms.api.IncrementalTransformOutput.write_dataframe()和partitionBy=['measure_date', 'measuring_time']实现您的目标。

检查 foundry docs了解更多。