Overview

When using Dataflow in Python, the official sample word countを読み解くためのメモ。

Will be added sequentially.

File IO

--Local execution --Reading & writing local files

import apache_beam as beam
import os


def first_pipeline():
    with beam.Pipeline('DirectRunner') as pipeline:
        #Read the file,"input_data"Assign to variable
        input_data = pipeline | 'ReadMyFile' >> beam.io.ReadFromText('./data/sample.txt')
        # input_Write the contents of data to a file
        input_data | beam.io.WriteToText('./data/output.txt')


if __name__ == '__main__':
    #Change working directory to the location of the executable
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    first_pipeline()

ParDo

Specify additional processes in sequence.


import apache_beam as beam
import os

def run_pipeline():
    with beam.Pipeline('DirectRunner') as pipeline:

        # [Final Output PCollection] = ([Initial Input PCollection]
        #                               | [First Transform]
        #                               | [Second Transform]
        #                               | [Third Transform])
        input_data = (pipeline
        | 'read_file' >> beam.io.ReadFromText('./data/sample.txt')
        | 'add_hoge_string' >> beam.ParDo(lambda line: line + "hoge")
        )

        # input_Write the contents of data to a file
        input_data | beam.io.WriteToText('./data/output.txt')


if __name__ == '__main__':
    #Change working directory to the location of the executable
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    run_pipeline()

Contents of sample.txt

Hello World
foo
bar

output

H
e
l
l
o
 
W
o
r
l
d
h
o
g
e
f
o
o
h
o
g
e
b
a
r
h
o
g
e

Hoge is added at the end of each line, but for some reason a line break is also added

Dataflow Primer wordcount and earlier

Overview

File IO