Diese Nachricht wurde von unseren Experten getestet, um die Richtigkeit dieses Essays sicherzustellen.
Lösung:
Ich habe versucht, eine Lösung mit dem oben zitierten Fall zu implementieren. Dort, sowie in anderen Ansätzen wie diesem, erhalten sie auch eine Liste von Dateinamen, laden aber alle Dateien in ein einziges Element, was bei großen Dateien nicht gut skalieren könnte. Deshalb habe ich mir überlegt, den Dateinamen zu jedem Datensatz hinzuzufügen.
Als Eingabe habe ich zwei csv-Dateien verwendet:
$ gsutil cat gs://$BUCKET/countries1.csv
id,country
1,sweden
2,spain
gsutil cat gs://$BUCKET/countries2.csv
id,country
3,italy
4,france
Mit . GCSFileSystem.match können wir auf metadata_list zugreifen, um FileMetadata abzurufen, die den Dateipfad und die Größe in Bytes enthalten. In meinem Beispiel:
[FileMetadata(gs://BUCKET_NAME/countries1.csv, 29),
FileMetadata(gs://BUCKET_NAME/countries2.csv, 29)]
Der Code lautet:
result = [m.metadata_list for m in gcs.match(['gs://{}/countries*'.format(BUCKET)])]
Wir werden jede der passenden Dateien in eine andere PCollection einlesen. Da wir die Anzahl der Dateien nicht im Voraus kennen, müssen wir programmatisch eine Liste von Namen für jede PCollection erstellen (p0, p1, ..., pN-1) erstellen und sicherstellen, dass wir für jeden Schritt eindeutige Bezeichnungen haben ('Read file 0', 'Read file 1', etc.):
variables = ['p{}'.format(i) for i in range(len(result))]
read_labels = ['Read file {}'.format(i) for i in range(len(result))]
add_filename_labels = ['Add filename {}'.format(i) for i in range(len(result))]
Dann lesen wir jede einzelne Datei in ihre entsprechende PCollection mit ReadFromText und dann rufen wir die AddFilenamesFn ParDo auf, um jeden Datensatz mit dem Dateinamen zu verknüpfen.
for i in range(len(result)):
globals()[variables[i]] = p | read_labels[i] >> ReadFromText(result[i].path) | add_filename_labels[i] >> beam.ParDo(AddFilenamesFn(), result[i].path)
wobei AddFilenamesFn ist:
class AddFilenamesFn(beam.DoFn):
"""ParDo to output a dict with filename and row"""
def process(self, element, file_path):
file_name = file_path.split("/")[-1]
yield {'filename':file_name, 'row':element}
Mein erster Ansatz war die direkte Verwendung einer Map-Funktion, was zu einem einfacheren Code führt. Allerdings, result[i].path wurde am Ende der Schleife aufgelöst und jeder Datensatz wurde fälschlicherweise der letzten Datei der Liste zugeordnet:
globals()[variables[i]] = p | read_labels[i] >> ReadFromText(result[i].path) | add_filename_labels[i] >> beam.Map(lambda elem: (result[i].path, elem))
Schließlich werden alle PCollections zu einer einzigen zusammengefasst:
merged = [globals()[variables[i]] for i in range(len(result))] | 'Flatten PCollections' >> beam.Flatten()
und überprüfen die Ergebnisse durch Protokollierung der Elemente:
INFO:root:{'filename': u'countries2.csv', 'row': u'id,country'}
INFO:root:{'filename': u'countries2.csv', 'row': u'3,italy'}
INFO:root:{'filename': u'countries2.csv', 'row': u'4,france'}
INFO:root:{'filename': u'countries1.csv', 'row': u'id,country'}
INFO:root:{'filename': u'countries1.csv', 'row': u'1,sweden'}
INFO:root:{'filename': u'countries1.csv', 'row': u'2,spain'}
Ich habe dies sowohl mit DirectRunner und DataflowRunner für Python SDK 2.8.0.
Ich hoffe, damit ist das Hauptproblem angesprochen und Sie können jetzt mit der Integration von BigQuery in Ihren vollständigen Anwendungsfall fortfahren. Möglicherweise müssen Sie dafür die Python Client Library verwenden, ich habe ein ähnliches Java-Beispiel geschrieben.
Vollständiger Code:
import argparse, logging
from operator import add
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io import ReadFromText
from apache_beam.io.filesystem import FileMetadata
from apache_beam.io.filesystem import FileSystem
from apache_beam.io.gcp.gcsfilesystem import GCSFileSystem
class GCSFileReader:
"""Helper class to read gcs files"""
def __init__(self, gcs):
self.gcs = gcs
class AddFilenamesFn(beam.DoFn):
"""ParDo to output a dict with filename and row"""
def process(self, element, file_path):
file_name = file_path.split("/")[-1]
# yield (file_name, element) # use this to return a tuple instead
yield {'filename':file_name, 'row':element}
# just logging output to visualize results
def write_res(element):
logging.info(element)
return element
def run(argv=None):
parser = argparse.ArgumentParser()
known_args, pipeline_args = parser.parse_known_args(argv)
p = beam.Pipeline(options=PipelineOptions(pipeline_args))
gcs = GCSFileSystem(PipelineOptions(pipeline_args))
gcs_reader = GCSFileReader(gcs)
# in my case I am looking for files that start with 'countries'
BUCKET='BUCKET_NAME'
result = [m.metadata_list for m in gcs.match(['gs://{}/countries*'.format(BUCKET)])]
result = reduce(add, result)
# create each input PCollection name and unique step labels
variables = ['p{}'.format(i) for i in range(len(result))]
read_labels = ['Read file {}'.format(i) for i in range(len(result))]
add_filename_labels = ['Add filename {}'.format(i) for i in range(len(result))]
# load each input file into a separate PCollection and add filename to each row
for i in range(len(result)):
# globals()[variables[i]] = p | read_labels[i] >> ReadFromText(result[i].path) | add_filename_labels[i] >> beam.Map(lambda elem: (result[i].path, elem))
globals()[variables[i]] = p | read_labels[i] >> ReadFromText(result[i].path) | add_filename_labels[i] >> beam.ParDo(AddFilenamesFn(), result[i].path)
# flatten all PCollections into a single one
merged = [globals()[variables[i]] for i in range(len(result))] | 'Flatten PCollections' >> beam.Flatten() | 'Write results' >> beam.Map(write_res)
p.run()
if __name__ == '__main__':
run()