I have been working on Apache Beam for a couple of days. I wanted to quickly iterate on the application I am working and make sure the pipeline I am building is error free. In spark we can use sc.parallelise and when we apply some action we get the value that we can inspect.

我已经在阿帕奇光束上工作了几天。我想快速迭代我正在工作的应用程序，并确保我正在构建的管道是没有错误的。在Spark中，我们可以使用sc.parallise，当我们应用某个操作时，我们会得到可以检查的值。

Similarly when I was reading about Apache Beam, I found that we can create a PCollection and work with it using following syntax

类似地，当我在阅读有关Apache Beam的文章时，我发现我们可以创建一个PCollection并使用以下语法来使用它

with beam.Pipeline() as pipeline:
    lines = pipeline | beam.Create(["this is test", "this is another test"])
    word_count = (lines 
                  | "Word" >> beam.ParDo(lambda line: line.split(" "))
                  | "Pair of One" >> beam.Map(lambda w: (w, 1))
                  | "Group" >> beam.GroupByKey()
                  | "Count" >> beam.Map(lambda (w, o): (w, sum(o))))
    result = pipeline.run()

I actually wanted to print the result to console. But I couldn't find any documentation around it.

我其实想把结果打印到控制台上。但我在它周围找不到任何文件。

Is there a way to print the result to console instead of saving it to a file each time?

You don't need the temp list. In python 2.7 the following should be sufficient:

你不需要临时工名单。在python2.7中，以下内容应该足够了：

def print_row(row):
    print row

(pipeline 
    | ...
    | "print" >> beam.Map(print_row)
)

result = pipeline.run()
result.wait_until_finish()

In python 3.x, print is a function so the following is sufficient:

在python3.x中，print是一个函数，因此以下内容就足够了：

(pipeline 
    | ...
    | "print" >> beam.Map(print)
)

result = pipeline.run()
result.wait_until_finish()

After exploring furthermore and understanding how I can write testcases for my application I figure out the way to print the result to console. Please not that I am right now running everything to a single node machine and trying to understand functionality provided by apache beam and how can I adopt it without compromising industry best practices.

在进一步探索并了解如何为应用程序编写测试用例之后，我想出了将结果打印到控制台的方法。请注意，我现在正在一台单节点机器上运行所有东西，并试图了解由apache Beam提供的功能，以及如何才能在不影响行业最佳实践的情况下采用它。

So, here is my solution. At the very last stage of our pipeline we can introduce a map function that will print result to the console or accumulate the result in a variable later we can print the variable to see the value

所以，这是我的解决方案。在流水线的最后阶段，我们可以引入一个map函数，该函数将结果打印到控制台或将结果累积到变量中，稍后我们可以打印变量以查看值

import apache_beam as beam

# lets have a sample string
data = ["this is sample data", "this is yet another sample data"]

# create a pipeline
pipeline = beam.Pipeline()
counts = (pipeline | "create" >> beam.Create(data)
    | "split" >> beam.ParDo(lambda row: row.split(" "))
    | "pair" >> beam.Map(lambda w: (w, 1))
    | "group" >> beam.CombinePerKey(sum))

# lets collect our result with a map transformation into output array
output = []
def collect(row):
    output.append(row)
    return True

counts | "print" >> beam.Map(collect)

# Run the pipeline
result = pipeline.run()

# lets wait until result a available
result.wait_until_finish()

# print the output
print output

Maybe logging info instead of print?

也许记录信息而不是打印？

def _logging(elem):
    logging.info(elem)
    return elem

P | "logging info" >> beam.Map(_logging)

Follow an example from pycharm Edu

效仿来自pycharm edu的例子

import apache_beam as beam

class LogElements(beam.PTransform):
    class _LoggingFn(beam.DoFn):

        def __init__(self, prefix=''):
            super(LogElements._LoggingFn, self).__init__()
            self.prefix = prefix

        def process(self, element, **kwargs):
            print self.prefix + str(element)
            yield element

    def __init__(self, label=None, prefix=''):
        super(LogElements, self).__init__(label)
        self.prefix = prefix

    def expand(self, input):
        input | beam.ParDo(self._LoggingFn(self.prefix))

class MultiplyByTenDoFn(beam.DoFn):

    def process(self, element):
        yield element * 10

p = beam.Pipeline()

(p | beam.Create([1, 2, 3, 4, 5])
   | beam.ParDo(MultiplyByTenDoFn())
   | LogElements())

p.run()

Output

输出

10
20
30
40
50
Out[10]: <apache_beam.runners.portability.fn_api_runner.RunnerResult at 0x7ff41418a210>

with beam.Pipeline() as pipeline:
lines = pipeline | beam.Create(["this is test", "this is another test"])
word_count = (lines 
              | "Word" >> beam.ParDo(lambda line: line.split(" "))
              | "Pair of One" >> beam.Map(lambda w: (w, 1))
              | "Group" >> beam.GroupByKey()
              | "Count" >> beam.Map(lambda o: (o[0],str(sum(o[1])))))
word_count | beam.ParDo(lambda x: print(x))
result = pipeline.run()

This works like a charm !

这就像一种护身符！

Output:
('this', '2')
('is', '2')
('test', '2')
('another', '1')
('this', '2')
('is', '2')
('test', '2')
('another', '1')

输出：(‘This’，‘2’)(‘is’，‘2’)(‘test’，‘2’)(‘Another’，‘1’)(‘This’，‘2’)(‘is’，‘2’)(‘test’，‘2’)(‘Another’，‘1’)

I know it isn't what you asked for but why don't you store it to a text file? It's always better than printing it via stdout and it isn't volatile

我知道这不是你想要的，但你为什么不把它存储到文本文件中呢？它总是比通过标准输出打印它更好，而且它不会挥发