重命名 Apache Beam 中的 BigQuery 列

Question

我尝试在 Python 中重命名 Apache Beam 管道中的 bigquery 行，如下例所示： 拥有 1 个包含完整数据的 PCollection 和 1 个仅包含 3 个字段的 PCollection，在 col1.2 中重命名为 col1，在 col2.2 中重命名为 col2...

如何正确应用我的过滤器来获得带有重命名行的第二个 PCollection？

def is_filtered(row):
    row['col1'] == row['col1.2']
    row['col2'] == row['col2.2']
    row['col3'] == row['col3.2']
    yield row


with beam.Pipeline() as pipeline:
    query = open('query.sql', 'r')
    bq_source = beam.io.BigQuerySource(query=query.read(),
                                       use_standard_sql=True)    
    main_table = \
        pipeline \
        | 'ReadBQData' >> beam.io.Read(bq_source) \

    cycle_table = (
        pipeline 
        | 'FilterMainTable' >> beam.Filter(is_filtered, main_table))

我也想过使用分区，但我发现的分区示例更多是关于对行的内容进行分区，而不是对行本身进行分区

Answer 1

Filter 运算符用于创建从源中删除行的 PCollection（预期返回布尔值）。如果要创建行以 1:1 转换的 PCollection，请使用 Map 运算符。这是一个例子：

def filter_columns(row):
    return {'col1.2': row['col1'],
            'col2.2': row['col2'],
            'col3.2': row['col3']}


with beam.Pipeline() as pipeline:
    query = open('query.sql', 'r')
    bq_source = beam.io.BigQuerySource(query=query.read(),
                                       use_standard_sql=True)    
    main_table = \
        pipeline \
        | 'ReadBQData' >> beam.io.Read(bq_source)

    cycle_table = (
        main_table 
        | 'FilterMainTable' >> beam.Map(filter_columns))