使用谷歌云数据流读取多个镶木地板文件时如何判断记录来自哪个文件答案

【问题标题】：How to tell which file a record came from when reading multiple parquet files with google cloud dataflow使用谷歌云数据流读取多个镶木地板文件时如何判断记录来自哪个文件
【发布时间】：2019-09-18 20:41:32
【问题描述】：

我需要能够将谱系追溯到单个 parquet 文件，并能够执行批量加载，例如，如果要在数据流中发现缺陷，则重播几年的 parquet 文件。

经过多次尝试，以下内容适用于批量加载，其中 options.input 是 RuntimeValueProvider 和 SplitFn 只是 yields str.split()：

with beam.Pipeline(options=PipelineOptions(), argv=args) as p:
    mainPipe = p \
        | 'CSV of URIs' >> beam.Create([options.input]) \
        | 'Split URIs into records' >> beam.ParDo(SplitFn(',')) \
        | "Read Parquet" >> beam.io.parquetio.ReadAllFromParquet(columns=[k for k in fields.keys()])

很遗憾，beam.io.parquetio.ReadAllFromParquet 不会说出每条记录来自哪个文件，ReadFromParquet、parquetio 的唯一其他功能也不会。

没有离开 Google Cloud Dataflow 或教团队 Java，我能做些什么将许多 parquet 文件一次加载到 BigQuery 并知道每条记录来自哪个文件？

【问题讨论】：

标签： python google-cloud-dataflow apache-beam parquet

【解决方案1】：

鉴于当前的 API，我没有看到针对此的预制解决方案。虽然您可以通过以下任一方式解决问题：

扩展/修改 ReadAllFromParquet 以将文件名附加到输出。
使用BQ tools 从镶木地板导入。不过，我不确定他们是否有完全相同的场景。

【讨论】：

按照您的建议，我制作了parquetio.py 的本地副本并将file_name 添加到字典row 的第254 行上方。 row['file_name'] = file_name。在概念上并不漂亮，但简单且有效。谢谢。