beam/dataflow ReadAllFromParquet没有读取任何内容，但我的工作仍然成功？

Question

我有一个数据流工作：

1. 从GCS读取包含其他文件名的文本文件
2. 将文件名传递给ReadAllFromParquet以读取。Parquet文件
3. 写入BigQuery

尽管我的工作“成功”了，但它基本上没有经过ReadAllFromParquet步骤的输出集合。我成功地读取了列表中的文件，例如:['gs://my_bucket/my_file1.snappy.parquet'、'gs://my_bucket/my_file2.snappy.parquet'、'gs://my_bucket/my_file3.snappy.parquet']我还在readallfromparquet之前的步骤中使用记录器确认此列表和文件的GCS路径是正确的。

 with beam.Pipeline(options=pipeline_options_batch) as pipeline_2:

    try:

        final_data = (
        pipeline_2
        |'Create empty PCollection' >> beam.Create([None])
        |'Get accepted batch file: {}'.format(runtime_options.complete_batch) >> beam.ParDo(OutputValueProviderFn(runtime_options.complete_batch))
        |'Read all filenames into a list'>> beam.ParDo(FileIterator(runtime_options.files_bucket))
        |'Read all files' >> beam.io.ReadAllFromParquet(columns=['locationItemId','deviceId','timestamp'])
        |'Process all files' >> beam.ParDo(ProcessSch2())
        |'Transform to rows' >> beam.ParDo(BlisDictSch2())
        |'Write to BigQuery' >> beam.io.WriteToBigQuery(
                table = runtime_options.comp_table, 
                schema = SCHEMA_2,
                project = pipeline_options_batch.view_as(GoogleCloudOptions).project, #options.display_data()['project'],
                create_disposition = beam.io.BigQueryDisposition.CREATE_IF_NEEDED,  #'CREATE_IF_NEEDED',#create if does not exist.
                write_disposition = beam.io.BigQueryDisposition.WRITE_APPEND    #'WRITE_APPEND' #add to existing rows,partitoning
                )
        )
    except Exception as exception:
        logging.error(exception)
        pass

这就是我的工作图表后面的样子：

有人知道这里可能出了什么问题吗？最好的调试方法是什么？我目前的想法是：

>

桶权限问题。我注意到我正在阅读的桶是奇怪的，因为早些时候我不能下载文件，尽管我是一个项目所有者。project的所有者只有“存储遗留桶所有者”。我添加了“存储管理”，然后在用自己的帐户手动下载文件时，它工作得很好。根据Dataflow文档，我已经确保默认的compute service帐户和Dataflow帐户在此bucket上都有“Storage Admin”。然而，也许这都是一个转移注意力的问题，因为最终如果有权限问题，我应该在日志中看到这一点，作业将失败？

特别注意

Read all files/ReadAllFiles/ReadRange.out0

VS

Read all files/Read all files/ReadRange.out0

路径的第一部分是两个作业的步骤的名称。但我认为第二个是apache_beam.io.fileBasedSource(https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/filebasedsource.py)中的ReadAllFiles类，ReadAllFromText和ReadAllFromParquet都调用它。

我想知道如果这是一个实验功能，有没有办法打开它？

Answer 1

您没有提到如果您使用的是最后一个Beam SDK，请尝试使用SDK2.16来测试最后的更改。

doc声明ReadAllFromParquet是一种实验性的功能，也是ReadFromParquet；尽管如此，ReadFromParquet仍被报告在以下线程中工作：从嵌套HDFS目录中读取parquet文件，您可能希望尝试使用此功能。