[PYTHON] Dataflow Primer wordcount et versions antérieures

Aperçu

Lors de l'utilisation de Dataflow en Python, l'exemple officiel nombre de motsを読み解くためのメモ。

Sera ajouté séquentiellement.

Fichier IO

--Exécution locale

• Lecture et écriture de fichiers locaux

import apache_beam as beam
import os


def first_pipeline():
    with beam.Pipeline('DirectRunner') as pipeline:
        #Lisez le fichier,"input_data"Attribuer à la variable
        input_data = pipeline | 'ReadMyFile' >> beam.io.ReadFromText('./data/sample.txt')
        # input_Ecrire le contenu des données dans un fichier
        input_data | beam.io.WriteToText('./data/output.txt')


if __name__ == '__main__':
    #Remplacez le répertoire de travail par l'emplacement du fichier exécutable
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    first_pipeline()

ParDo

Spécifiez les processus supplémentaires en séquence.

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_Ecrire le contenu des données dans un fichier
        input_data | beam.io.WriteToText('./data/output.txt')


if __name__ == '__main__':
    #Remplacez le répertoire de travail par l'emplacement du fichier exécutable
    os.chdir(os.path.dirname(os.path.abspath(__file__)))
    run_pipeline()

Contenu de sample.txt

Hello World
foo
bar

production

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 est ajouté à la fin de chaque ligne, mais pour une raison quelconque, un saut de ligne est également ajouté