'FlatMap 只能用于可调用对象...'答案

【问题标题】：'FlatMap can be used only with callable objects...''FlatMap 只能用于可调用对象...'
【发布时间】：2018-08-28 22:44:25
【问题描述】：

我的代码已附上。我需要读入两个 CSV。我读入第一个 CSV，然后我想将该 PCollection 作为侧输入传递给我将逐行读取的另一个 CSV 文件。然后我想产生连接到 FlatMap 函数的两个元素。问题是，我无法将数据传递给函数（我使用的是 Python）。我在网上查看了很多示例，其他人在早期版本中已经这样做了。我知道它实际上在做某事，因为我至少可以将左侧的 csv 写入文本文件，并且可以看到它将每一行更改为一个键值对。非常感谢您的帮助，感谢您的阅读。

from __future__ import absolute_import
import logging
import csv
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class append_lr(beam.DoFn):
    def __init__(self, lineup):
        self._lineup=(1,2)

    def process(self, left, right):
        bla=left
        burp=right
        both=left+right
        yield both


class MyCsvFileSource(beam.io.filebasedsource.FileBasedSource):
    def read_records(self, file_name, range_tracker):
        self._file = self.open_file(file_name)
        reader = csv.DictReader(self._file)
        for rec in reader:
            yield rec

def combine_lines():
    with beam.Pipeline(options=PipelineOptions()) as p:

        left_side = p | 'Read_Left_Side' >> beam.io.Read(MyCsvFileSource('/folder/left_side.csv'))
        left_and_right = (p | 'Read_Rght_Side' >> beam.io.Read(MyCsvFileSource('/folder/right_side.csv'))
                     | beam.FlatMap(append_lr, beam.pvalue.AsIter(left_side)))
        left_and_right | 'Write' >> beam.io.WriteToText('/folder/', file_name_suffix='test_output.csv')

def run(argv=None):
    combine_lines()

if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run(None)

【问题讨论】：

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

【解决方案1】：

原来我应该将 DoFn 实现为“可调用”对象，下面的 Apache Beam 代码文档 sn-p 说：

def FlatMap(fn, *args, **kwargs):  # pylint: disable=invalid-name
""":func:FlatMap 类似于 :class: ParDo 只是它需要一个 callable 来指定转换。

所以我将函数从一个类更改为 def，就像一个魅力。其他一些代码更改也显示了如何基本上构建一个带有来自另一个 PCollection 的侧面输入的 for 循环（注意，如果您在本地运行它，请确保您的左右两侧的测试文件相当小，这会产生巨大的尺寸输出！）：

from __future__ import absolute_import
import logging
import csv
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class MyCsvFileSource(beam.io.filebasedsource.FileBasedSource):
    def read_records(self, file_name, range_tracker):
        self._file = self.open_file(file_name)
        reader = csv.DictReader(self._file)
        for rec in reader:
            yield rec

def append_lr(left_, right_):
    for thingy in right_:
        yield (left_, thingy)

def combine_sides():
    with beam.Pipeline(options=PipelineOptions()) as p:
        left_file = '/path/to/file/left.csv'
        right_file = '/path/to/file/right.csv'
        test_output = '/path/to/file/outputs/'

        left_side = p | 'Read_Left_Side' >> beam.io.Read(MyCsvFileSource(left_file))
        right_side = p | 'Read_Right_Side' >> beam.io.Read(MyCsvFileSource(right_file))
        all_combos = left_side | beam.FlatMap(append_lr, beam.pvalue.AsIter(right_side))
        all_combos | 'Write' >> beam.io.WriteToText(test_output, file_name_suffix='purple_nurple.csv')

def run(argv=None):
    combine_sides()

if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run(None)

【讨论】：