如何将扫描仪中的投影列合并到新的数据集分区中答案

【问题标题】：How to incorporate projected columns in scanner into new dataset partitioning如何将扫描仪中的投影列合并到新的数据集分区中
【发布时间】：2022-11-04 16:43:12
【问题描述】：

假设我加载了一个数据集

myds=ds.dataset('mypath', format='parquet', partitioning='hive')
myds.schema
# On/Off_Peak: string
# area: string
# price: decimal128(8, 4)
# date: date32[day]
# hourbegin: int32
# hourend: int32
# inflation: string rename to Inflation
# Price_Type: string
# Reference_Year: int32
# Case: string
# region: string rename to Region

我的最终目标是使用以下投影重新保存数据集：

projection={'Region':ds.field('region'),
            'Date':ds.field('date'),
            'isPeak':pc.equal(ds.field('On/Off_Peak'),ds.scalar('On')),
            'Hourbegin':ds.field('hourbegin'),
            'Hourend':ds.field('hourend'),
            'Inflation':ds.field('inflation'),
            'Price_Type':ds.field('Price_Type'),
            'Area':ds.field('area'),
            'Price':ds.field('price'),
            'Reference_Year':ds.field('Reference_Year'),
            'Case':ds.field('Case'),
            }

我做扫描仪

scanner=myds.scanner(columns=projection)

现在我尝试保存我的新数据集

ds.write_dataset(scanner, 'newpath',
                partitioning=['Reference_Year', 'Case', 'Region'], partitioning_flavor='hive',
                format='parquet')

但我明白了

KeyError: 'Column Region does not exist in schema'

我可以通过将partitioning 更改为['Reference_Year', 'Case', 'region'] 以匹配非投影列（然后更改所有这些目录的名称）来解决此问题，但是有没有办法直接做到这一点？

假设我的分区需要计算不仅仅是更改列名。我是否必须一步保存非分区数据集才能获取新列，然后再执行另一个保存操作来创建分区数据集？

【问题讨论】：

标签： pyarrow

【解决方案1】：

编辑：此错误已在 pyarrow 10.0.0 中修复

对我来说它看起来像一个错误。就好像write_dataset 正在查看dataset_schema 而不是projected_schema

我认为您可以通过在扫描仪上调用to_reader 来解决它。

table = pa.Table.from_arrays(
    [
        pa.array(['a', 'b', 'c'], pa.string()),
        pa.array(['a', 'b', 'c'], pa.string()),
    ],
    names=['region', "Other"]
)
table_dataset = ds.dataset(table)
columns={
    "Region": ds.field('region'),
    "Other": ds.field('Other'),
}
scanner = table_dataset.scanner(columns=columns)

ds.write_dataset(
    scanner.to_reader(), 
    'newpath',
    partitioning=['Region'], partitioning_flavor='hive',
    format='parquet')

我已经报告了这个问题here

【讨论】：