Add file name and timestamp into each record in BigQuery using Dataflow

user_14376281 0
Viewed 858

I have a few .txt files with data in JSON to be loaded to google BigQuery table. Along with the columns in the text files I will need to insert filename and current timestamp for each rows. It is in GCP Dataflow with Python 3.7

I accessed the Filemetadata containing the filepath and size using GCSFileSystem.match and metadata_list.

I believe I need to get the pipeline code to run in a loop, pass the filepath to ReadFromText, and call a FileNameReadFunction ParDo.

   (p
        | "read from file" >> ReadFromText(known_args.input)
        | "parse" >> beam.Map(json.loads)
        | "Add FileName" >>  beam.ParDo(AddFilenamesFn(), GCSFilePath)
        | "WriteToBigQuery" >> beam.io.WriteToBigQuery(known_args.output,          
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
     )

I followed the steps in Dataflow/apache beam - how to access current filename when passing in pattern? but I can't make it quite work.

Any help is appreciated.

apache-beam google-cloud-dataflow google-bigquery
1 Answers

You can use textio.ReadFromTextWithFilename instead of ReadFromText. That will produce a PCollection of (filename,line) tuples.

To include the file and timestamp in your output json record, you could change your "parse" line to

| "parse" >> beam.map(lambda (file, line): {
    **json.loads(line),
    "filename": file,
    "timestamp": datetime.now()})
user_48481130
Related