Streaming events to Big Query - Dataflow - Best way to insert **epoch timestamp** (int) to timestamp column

user_16976041 0
Viewed 362

I am trying to write events in streaming into Big Query table using dataflow apache beam (python) having a problem with timestamp format.

I have an event (json) with epoch timestamp (int) values which I want to insert into BIG QUERY table with timestamp columns .

What is the best way to do it ? Can I do it without parsing each event ? can I declare the format of the timestamp received ?

for example :

event= {'ts' : 1630494181342 ,'user' : 'anat'}

Into table :

{ts: timestamp , user: string}
bq apache-beam epoch timestamp python
1 Answers

Apache Beam applies the transformations on all the elements that are passed through the pipeline. So, the defined pipeline steps will convert the epoch time epoch time to datetime format on a per element basis before ingesting it into BigQuery.

The following Apache Beam pipeline code will convert the epoch time in seconds to datetime ie. timestamp

Example code:

import apache_beam as beam
 
class GetTimestamp(beam.DoFn):
 def process(self, mytime, timestamp=beam.DoFn.TimestampParam):
   yield '{}'.format(timestamp.to_utc_datetime())
 
with beam.Pipeline() as pipeline:
 plant_timestamps = (
     pipeline
     | 'My Time' >> beam.Create([
         {'ts': 1633013727, 'user': 'anant'},
         {'ts': 1590969600, 'user':'samwilliam'},   
     ])
     | 'With timestamps' >> beam.Map(
         lambda mytime: beam.window.TimestampedValue(mytime, mytime['ts']))
     | 'Get timestamp' >> beam.ParDo(GetTimestamp())
     | beam.Map(print)
 )
user_157504730
Related