Correct format to sink to BigQuery using Dataflow Apache Beam

Question

It seems like an error occur in bcdate, and I transform it to correct format but the error is still here, my code is below:

def transform_pandas(data):
    import pandas as pd
    import json
    import datetime as dt

    df = pd.DataFrame([data])
    
    # Fill all columns with null if there is no data    
    columns = ['peakid', 'route1', 'bcdate', 'pkname', 'heightm']
    df = df.reindex(columns, fill_value='null', axis=1)
    
    df['bcdate'] = pd.to_datetime(df['bcdate'], errors='coerce').dt.strftime('%Y-%m-%d')
    
    return json.loads(df.to_json(orient = 'records'))

and code to write to BigQuery:

output = (
        (input_p1, input_p2)
        | 'Join' >> beam.CoGroupByKey()
        | 'Final Dict' >> beam.Map(lambda el: to_final_dict(el[1])) >> it got a result here
        | 'Transformation' >> beam.Map(transform_pandas) >> the error happen in here
        | beam.Map(print)
        | 'Write To BigQuery' >> beam.io.gcp.bigquery.WriteToBigQuery(
           table='project:dataset.expeditions',
           # schema='peakid:STRING,route1:STRING,bcdate:DATETIME,pkname:STRING,heightm:INTEGER',
           method='FILE_LOADS',
           custom_gcs_temp_location='gs://bucket/folder/temp',
           create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
           write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE)    
    )

Here is my result

[{'peakid': 'ACHN', 'route1': '', 'bcdate': None, 'pkname': 'Aichyn', 'heightm': '6055'}]
[{'peakid': 'AGLE', 'route1': 'null', 'bcdate': None, 'pkname': 'Agole East', 'heightm': '6675'}]
[{'peakid': 'KCHS', 'route1': 'NW Ridge', 'bcdate': '2019-10-24', 'pkname': 'Kangchung Shar', 'heightm': '6063'}]
[{'peakid': 'LNAK', 'route1': 'SSE Ridge', 'bcdate': '2015-09-17', 'pkname': 'Lhonak', 'heightm': '6070'}]
[{'peakid': 'SPH1', 'route1': 'S Face', 'bcdate': '2017-04-14', 'pkname': 'Sharphu I', 'heightm': '6433'}]
...

But I got the error when I'm trying to sink to BigQuery: BigQuery job beam_bq_job_LOAD_AUTOMATIC_JOB_NAME_LOAD_STEP_329_215864ba592a2e01f0c4e2157cc60c47_3a904aab56c3444bb56bda650a7404b3 failed. Error Result: [while running 'Write To BigQuery/BigQueryBatchFileLoads/WaitForDestinationLoadJobs']

Mazlum Tosun · Answer

Your bcdate has the following date format YYYY-MM-DD in the result Dict but in your Bigquery schema and table you have a datetime type.

You can the check the doc to pass the correct format for datetime : https://cloud.google.com/bigquery/docs/reference/standard-sql/datetime_functions?hl=en

For example : 2017-05-26T00:00:00

You can also find the info on the beam documentation :

https://beam.apache.org/documentation/io/built-in/google-bigquery/

Example :

bigquery_data = [{     
'string': 'abc',     
'bytes': base64.b64encode(b'\xab\xac'),     
'integer': 5,     
'float': 0.5,     
'numeric': Decimal('5'),     
'boolean': True,     
'timestamp': '2018-12-31 12:44:31.744957 UTC',     
'date': '2018-12-31',     
'time': '12:44:31',     
'datetime': '2018-12-31T12:44:31',     
'geography': 'POINT(30 10)'  
}]

Correct format to sink to BigQuery using Dataflow Apache Beam

1 Answers