apache beam如何创建支持模式的PCollection

Question

我试图使用apache beam读取两个数据集，如果第二个数据集中存在与ID匹配的行，则更新第一个数据集。以下是梁管道：

class MapDataFrameRow(beam.DoFn):

    def process(self, element):
        key = element.id
        yield key, element


class MergeTuples(beam.DoFn):

    def process(self, element):
        key, data = element
        if len(data[1]) > 0:
            yield data[1][0]
        else:
            yield data[0][0]

def run(pipeline_options, state_file, gcs_input_file_path, gcs_output_path):
    with beam.Pipeline(options=pipeline_options) as pipeline:

        current_state_df = pipeline | 'Read current state CSV' >> read_csv(state_file)
        updates_df = pipeline | 'Read data updates CSV' >> read_csv(gcs_input_file_path)

        # Translate beam dataframe to pcollections
        #  MapDataFrameRow is a DoFn that map the ID as the tuple key.
        cs_pc = to_pcollection(current_state_df, include_indexes=False) | "Map #1 to KV" >> beam.ParDo(MapDataFrameRow())
        nu_pc = to_pcollection(updates_df, include_indexes=False) | "Map #2 to KV" >> beam.ParDo(MapDataFrameRow())

        merged = (
                (cs_pc, nu_pc)
                | 'group by key' >> beam.CoGroupByKey()
                | 'keep most recent' >> beam.ParDo(MergeTuples())
                # | 'To rows' >> beam.Map(lambda word: beam.Row(??))
                | beam.ParDo(print)
        )


        df = to_dataframe(merged)
        df.to_csv(gcs_output_path)

代码产生以下输出：

BeamSchema_c46d9a9f_a000_453e_87fe_7f4b372c21a1(id=143, area=130.0, .... )
BeamSchema_c46d9a9f_a000_453e_87fe_7f4b372c21a1(id=144, area=130.0, ... )
BeamSchema_c46d9a9f_a000_453e_87fe_7f4b372c21a1(id=145, area=130.0, ... )
...

遗憾的是，我无法将合并的pcollection映射到数据帧中，出现以下错误：

TypeError: Could not determine schema for type hint Any. Did you mean to create a schema-aware PCollection? See https://s.apache.org/beam-python-schemas

我已尝试使用“到行”

编辑：问题似乎是to_dataframe（合并）不喜欢使用由两个to_pcollections生成的两个不同的BeamSchema_c46d9a9f_xxxx和BeamSchema_3369b0_xxxx。换句话说，我正在创建两个单独的p集合，然后我从一个源或另一个源中选择，然后尝试创建一个新的数据帧，但是因为它们看起来像是单独的模式，他不喜欢它。合并两个数据帧的方法是什么？

Answer 1

首先，我看到您应用了变换光束。ParDo（打印）以生成合并的PCollection。这意味着您尝试转换的PCollection将包含此函数返回的对象。是否使用自定义打印功能在打印后返回原始对象？

如果上面没有解决问题，您必须确保Beam Python可以从PCollection表示的对象类型推断Schema。对于自定义类型，这可以通过将对象转换为NamedTuple来完成。有关更多详细信息，请参阅此处。

这应该允许to_dataframe正确发现PCollection包含的对象类型的模式。