输出类型中beam.ParDo 和beam.Map 的区别？

Question

我正在使用 Apache-Beam 运行一些数据转换，包括从 txt、csv 和不同数据源中提取数据。我注意到的一件事是使用beam.Map和beam.ParDo时的结果差异

在下一个示例中：

我正在读取 csv 数据，在第一种情况下，使用beam.ParDo将其传递给DoFn，它提取第一个元素，即日期，然后打印它。第二种情况，我直接用beam.Map做同样的事情，然后打印出来。

class Printer(beam.DoFn):
    def process(self,data_item):
        print data_item

class DateExtractor(beam.DoFn):
    def process(self,data_item):
        return (str(data_item).split(','))[0]

data_from_source = (p
                    | 'ReadMyFile 01' >> ReadFromText('./input/data.csv')
                    | 'Splitter using beam.ParDo 01' >> beam.ParDo(DateExtractor())
                    | 'Printer the data 01' >> beam.ParDo(Printer())
                    )

copy_of_the_data =  (p
                    | 'ReadMyFile 02' >> ReadFromText('./input/data.csv')
                    | 'Splitter using beam.Map 02' >> beam.Map(lambda record: (record.split(','))[0])
                    | 'Printer the data 02' >> beam.ParDo(Printer())
                    )

我在两个输出中注意到的是下一个：

##With beam.ParDo##
2
0
1
7
-
0
4
-
0
3
2
0
1
7

##With beam.Map##
2017-04-03
2017-04-03
2017-04-10
2017-04-10
2017-04-11
2017-04-12
2017-04-12

我觉得这很奇怪。我想知道打印功能是否有问题？但是在使用不同的转换后，它显示出相同的结果。例如运行：

| 'Group it 01' >> beam.Map(lambda record: (record, 1))

仍然返回相同的问题：

##With beam.ParDo##
('8', 1)
('2', 1)
('0', 1)
('1', 1)

##With beam.Map##
(u'2017-04-08', 1)
(u'2017-04-08', 1)
(u'2017-04-09', 1)
(u'2017-04-09', 1)

知道是什么原因吗？我在beam.Map和beam.ParDo之间的区别中缺少什么？？？

Answer 1

简答

您需要将 a 的返回值包装ParDo到一个列表中。

更长的版本

ParDos一般来说，可以为单个输入返回任意数量的输出，即对于单个输入字符串，您可以发出零个、一个或多个结果。出于这个原因，Beam SDK 将 a 的输出ParDo视为不是单个元素，而是元素的集合。

在您的情况下，ParDo发出单个字符串而不是集合。Beam Python SDK 仍然尝试将其输出解释ParDo为元素集合。它通过将您发出的字符串解释为字符集合来实现。因此，您ParDo现在可以有效地生成单个字符流，而不是字符串流。

您需要做的是将您的返回值包装到一个列表中：

class DateExtractor(beam.DoFn):
    def process(self,data_item):
        return [(str(data_item).split(','))[0]]

注意方括号。有关更多示例，请参阅编程指南。

Map，另一方面，可以被认为是 的特例ParDo。Map预计为每个输入产生一个输出。因此，在这种情况下，您只需从 lambda 返回一个值，它就会按预期工作。

而且您可能不需要将data_itemin包装起来str。根据文档，ReadFromText转换产生字符串。