Pandas：合并具有不同列 dtypes 的镶木地板文件 - 用预定义的模式编写镶木地板？

Question

我需要将非常大的数据库表导出到 s3。 我这样做是通过并行化 pandas read_sql（使用 processpool），并使用我的表的主键 id 生成一个范围来为每个工作人员选择的。 - 这会导致非常快速的导出。

process 1: id between 1 and 9   -> 1.pq
process 2: id between 10 and 19 -> 2.pq
process 3: id between 20 and 29 -> 3.pq

每个工作人员将生成的数据帧写入同一个文件夹。

问题在于我的数据：我拥有的某些列并不总是被填充（例如 Deleted？null vs 1） - 所以我的一些镶木地板将删除列数据类型为 null 其他为 Int64。

当我尝试从 pyarrow、fastparquet 或 pyspark 读取数据集时，我会收到有关架构的各种错误。

我曾尝试研究箭头表，但到目前为止只找到一种方法来定义仅用于验证的模式。

复制：

import pandas as pd
import pyarrow.parquet as pq

data=pd.DataFrame([[1,None],[1,None]])
data2=pd.DataFrame([[1,1],[1,1]])
data.columns = data.columns.astype(str) ## Parquet requires string column names
data2.columns = data2.columns.astype(str)
data.to_parquet('./outputs/1.pq')
data2.to_parquet('./outputs/2.pq')
pq.ParquetDataset('./outputs')

我希望它推断出我的列“1”是 int，但它会发生冲突。我尝试禁用 schema_validation，但这只是隐藏了问题，直到我实际处理它。

ValueError: Schema in ../outputs/2.pq was different. 
0: int64
1: int64
metadata
--------
{b'pandas': b'{"index_columns": [{"kind": "range", "name": null, "start": 0, "'
            b'stop": 2, "step": 1}], "column_indexes": [{"name": null, "field_'
            b'name": null, "pandas_type": "unicode", "numpy_type": "object", "'
            b'metadata": {"encoding": "UTF-8"}}], "columns": [{"name": "0", "f'
            b'ield_name": "0", "pandas_type": "int64", "numpy_type": "int64", '
            b'"metadata": null}, {"name": "1", "field_name": "1", "pandas_type'
            b'": "int64", "numpy_type": "int64", "metadata": null}], "creator"'
            b': {"library": "pyarrow", "version": "0.14.0"}, "pandas_version":'
            b' "0.24.2"}'}

vs

0: int64
1: null
metadata
--------
{b'pandas': b'{"index_columns": [{"kind": "range", "name": null, "start": 0, "'
            b'stop": 2, "step": 1}], "column_indexes": [{"name": null, "field_'
            b'name": null, "pandas_type": "unicode", "numpy_type": "object", "'
            b'metadata": {"encoding": "UTF-8"}}], "columns": [{"name": "0", "f'
            b'ield_name": "0", "pandas_type": "int64", "numpy_type": "int64", '
            b'"metadata": null}, {"name": "1", "field_name": "1", "pandas_type'
            b'": "empty", "numpy_type": "object", "metadata": null}], "creator'
            b'": {"library": "pyarrow", "version": "0.14.0"}, "pandas_version"'
            b': "0.24.2"}'}

Answer 1

您可以创建自己的自定义“pyarrow 架构”并使用您的架构投射每个 pyarrow 表。

  import pyarrow as pa
    import pyarrow.parquet as pq
    def merge_small_parquet_files(small_files, result_file):
        pqwriter = None
        for small_file in small_files:
            table = pq.read_table(small_file)
            pyarrow_schema = get_pyarrow_schema()
            if not pqwriter:
                pqwriter = pq.ParquetWriter(result_file,
                                        schema=pyarrow_schema,
                                        compression='GZIP',
                                        coerce_timestamps='ms', allow_truncated_timestamps=True)
                table = table.cast(pyarrow_schema)
                pqwriter.write_table(table)
                table = None
                del table
            if pqwriter:
                pqwriter.close()

    def get_pyarrow_schema():
        fields = []
        fields.append(pa.field('first_name', pa.string()))
        fields.append(pa.field('last_name', pa.string()))
        fields.append(pa.field('Id', pa.float64()))
        fields.append(pa.field('Salary', pa.float64()))
        fields.append(pa.field('Time', pa.timestamp('ms')))
        pyarrow_schema = pa.schema(fields)
        return pyarrow_schema
    if __name__ == '__main__':
        small_files = ['file1.parquet', 'file2.parquet', 'file3.parquet', 'file4.parquet']
        result_file = 'large.parquet'
        merge_small_parquet_files(small_files, result_file)