Is there a way to Keep track of all the bad records that are allowed while loading a ndjson file into Bigquery

user_13608103 0
Viewed 785

I have a requirement where I need to keep track of all the bad records that were not feeded into bigquery after allowing max_bad_records. So I need them written in a File on storage for Future reference. I'm using BQ API for Python, Is there a way we can achieve this? I think if we are allowing max_bad_records we dont have the details of failed loads in BQ Load Job.

Thanks

google-cloud-storage google-bigquery python-3.x
1 Answers

Currently, there isn't a direct way of accessing and saving the bad records. However, you can access some job statistics including the reason why the record was skipped within BigQuery _job_statistics().

I have created an example, in order to demonstrate how the statistics will be shown. I have the following sample .csv file in a GCS bucket: 

name,age
robert,25
felix,23
john,john

As you can see, the last row is a bad record, because I will import age as INT64 and there is a string in that row. In addition, I used the following code to upload it to BigQuery: 

from google.cloud import bigquery
client = bigquery.Client()
table_ref = client.dataset('dataset').table('table_name')

job_config = bigquery.LoadJobConfig(
        schema=[
            bigquery.SchemaField("name", "STRING"),
            bigquery.SchemaField("age", "INT64"),
            ]
        )
job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE
job_config.skip_leading_rows = 1
job_config.max_bad_records = 5
#job_config.autodetect = True
# The source format defaults to CSV, so the line below is optional.
job_config.source_format = bigquery.SourceFormat.CSV
uri = "gs://path/file.csv"
load_job = client.load_table_from_uri(
    uri, table_ref, job_config=job_config
)  # API request
print("Starting job {}".format(load_job.job_id))

load_job.result()  # Waits for table load to complete.
print("Job finished.")

destination_table = client.get_table(table_ref)
print("Loaded {} rows.".format(destination_table.num_rows))


#Below all the statistics that might be useful in your case

job_state = load_job.state
job_id = load_job.job_id
error_result = load_job.error_result
job_statistics = load_job._job_statistics()
badRecords = job_statistics['badRecords']
outputRows = job_statistics['outputRows']
inputFiles = job_statistics['inputFiles']
inputFileBytes = job_statistics['inputFileBytes']
outputBytes = job_statistics['outputBytes']

print("***************************** ")
print(" job_state:      " + str(job_state))
print(" non fatal error: " + str(load_job.errors))
print(" error_result:   " + str(error_result))
print(" job_id:         " + str(job_id))
print(" badRecords:     " + str(badRecords))
print(" outputRows:     " + str(outputRows))
print(" inputFiles:     " + str(inputFiles))
print(" inputFileBytes: " + str(inputFileBytes))
print(" outputBytes:    " + str(outputBytes))
print(" ***************************** ")

print("------ load_job.errors ")

The output from the statistics :

*****************************
 job_state:      DONE
 non fatal errors: [{u'reason': u'invalid', u'message': u"Error while reading data, error message: Could not parse 'john' as INT64 for field age (position 1) starting at location 23", u'location': u'gs://path/file.csv'}]
 error_result:   None
 job_id:         b2b63e39-a5fb-47df-b12b-41a835f5cf5a
 badRecords:     1
 outputRows:     2
 inputFiles:     1
 inputFileBytes: 33
 outputBytes:    26
 *****************************

As it is shown above, the erros field returns the non fatal errors, which includes the bad records. In other words, it retrieves individual errors generated by the job. Whereas, the error_result returns the error information as the job as a whole.

I believe these statistics might help you analyse your bad records. Lastly, you can output them into a file, using write(), such as: 

with open("errors.txt", "x") as f: 
    f.write(load_job.errors)
    f.close()
user_125713870
Related