Spark在Foundry中可以使用分区修剪吗

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

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

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

创建分区的代码如下：

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 没有使用任何分区，计划中的分区过滤器为空。

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 使用分区修剪？

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

检查铸造文档 https://www.palantir.com/docs/foundry/transforms-python/transforms-python-api-classes/#incrementaltransformoutput了解更多。