beam.ParDo和beam.Map在输出类型上的差异？

Question

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

在下一个样本中：

我正在读取csv数据，在第一种情况下，使用DoFn将其传递给beam.ParDo，后者提取第一个元素日期，然后打印它。在第二个例子中，我直接使用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

短答案

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

更长版本

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

在您的示例中，ParDo只发出一个字符串而不是一个集合。Beam Python仍然试图像解释元素集合一样解释ParDo的输出。它通过解释您作为字符集合发出的字符串来做到这一点。正因为如此，您的ParDo现在有效地生成单个字符流，而不是字符串流。

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

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

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

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

而且您可能不需要将data_item包装在str中。根据医生的说法 ReadFromText转换产生字符串。