Skip to content

Dataflow/apache beam - wie kann man auf den aktuellen Dateinamen zugreifen, wenn ein Muster übergeben wird?

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()



Nutzen Sie unsere Suchmaschine

Suche
Generic filters

Schreibe einen Kommentar

Deine E-Mail-Adresse wird nicht veröffentlicht.