有没有办法跟踪将 ndjson 文件加载到 Bigquery 时允许的所有不良记录

Question

我有一个要求，我需要跟踪所有在允许max_bad_records后未输入 bigquery 的不良记录。所以我需要将它们写在存储文件中以供将来参考。我正在使用 BQ API for Python，有没有办法实现这一点？我认为如果我们允许max_bad_records我们就没有 BQ 加载作业中失败加载的详细信息。

谢谢

Answer 1

目前，没有直接的方法来访问和保存不良记录。但是，您可以访问一些作业统计信息，包括在 BigQuery _job_statistics(( 中跳过记录的原因。

我创建了一个示例，以演示如何显示统计信息。我在 GCS 存储桶中有以下示例.csv文件：

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

如您所见，最后一行是一个糟糕的记录，因为我将年龄导入为INT64，并且该行中有一个字符串。此外，我使用以下代码将其上传到 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 ")

统计数据的输出：

*****************************
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
*****************************

如上所示，erros 字段返回非致命错误，其中包括错误记录。换句话说，它检索作业生成的单个错误。而error_result将错误信息作为整个作业返回。

我相信这些统计数据可能会帮助你分析你的不良记录。最后，您可以使用 write(( 将它们输出到文件中，例如：

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