From 75241865293e5d82c40efa650e16c174e4033a29 Mon Sep 17 00:00:00 2001 From: Nikhil Lobo <nikhil.lobo@niveussolutions.com> Date: Thu, 5 Dec 2024 14:51:55 +0530 Subject: [PATCH] beam assignment --- beamAst.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 beamAst.py diff --git a/beamAst.py b/beamAst.py new file mode 100644 index 0000000..c566dc8 --- /dev/null +++ b/beamAst.py @@ -0,0 +1,19 @@ +import apache_beam as beam + +table_schema = 'column1:STRING, column2:STRING' +table_spec = "[project_id]:[dataset_id].[table_id]" + + +with beam.Pipeline() as pipeline: + ip = ( pipeline + |beam.io.ReadFromText("gs://path-to-file",skip_header_lines=True) + |beam.Map(lambda x : x.split(",")) + |beam.Map(lambda y : {'column1': y[0],'column2': y[1]}) + |beam.io.WriteToBigQuery( + table_spec, + schema=table_schema, + write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE, + create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED) + + + ) -- GitLab