json - BigQuery 加载 JSON 错误 "Could not convert value to string"

Question

我正在尝试将 JSON 事件数据从 Google Cloud Storage 加载到 BigQuery，我想将所有内容都加载为字符串，然后再转换它们，因为对于某些消息，它们看起来像这样:

{"id":"123"}
{"id":234}

我写的schema.json是:

[
    {
        "name": "id",
        "type": "STRING",
        "mode": "NULLABLE"
    }
]

然后我使用 bq cli 加载它:

LOC="--location US"
INPUT=sample.json
SCHEMA=schema.json
bq $LOC load \
    --source_format=NEWLINE_DELIMITED_JSON \
    --ignore_unknown_values \
    --schema=$SCHEMA \
    nov2020.test \
    $INPUT

它会因为这个错误而失败:

Failure details:

• Error while reading data, error message: JSON processing encountered too many errors, giving up. Rows: 2; errors: 1; max bad: 0; errorpercent: 0
• Error while reading data, error message: JSON parsing error in row starting at position 13: Could not convert value to string. Field: id;Value: 234

我不想使用 --max_bad_records 跳过这些记录，我想通过不使用自动检测，我可以将所有内容都读取为字符串。

我每天要处理大约 80GB 的这些 JSON 文件，那么我该如何处理这个错误？我唯一的选择是在加载到 BigQuery 之前检查每条 JSON 消息并格式化 id 字段吗？

我找到了 another post saying to use a plug in for fluentd ，但我不确定这是否适用于此，因为我的数据不是使用 fluentd 创建的。

谢谢!

Answer 1

解决问题的最简单方法是用 Dataflow 作业替换加载。下面的代码基本上会读取存储桶中的文件，修复 json 记录，然后将修复的记录写入 BigQuery。

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.internal.clients import bigquery
import json

class FixJson(beam.DoFn):
    def __init__(self):
        beam.DoFn.__init__(self)

    def process(self, element, *args, **kwargs):
        row = str(element)
        json_obj = json.loads(row)
        json_obj["field"] = str(json_obj["field"])
        return [json_obj]


table_spec = bigquery.TableReference(
    projectId='<your-project>',
    datasetId='<your-dataset>',
    tableId='<your-table>')

p = beam.Pipeline(options=PipelineOptions())
p1 = p | "Read data from GCS" >> beam.io.ReadFromText('gs://<your-bucket>/*') \
        | "Fix json" >> beam.ParDo(FixJson())\
        | "Write to bq" >> beam.io.WriteToBigQuery(table_spec,
                                            custom_gcs_temp_location = '<some-temporary-bucket>',
                                            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)

p.run()

让我们在某些部分打破这个管道并逐步完成它:

1. 在管道的第一行，Dataflow 将以文本形式读取所提供存储桶中的所有文件。您应该更改此设置并放置正确的路径以匹配必须在给定日期插入 BigQuery 的数据。

   “从GCS读取数据”>> beam.io.ReadFromText('gs:///*')

2. 在第二步中，Dataflow 将在 ParDo 转换中使用 FixJson 函数来更改 json 的结构。您应该根据数据的复杂程度以及需要更改的程度来更改逻辑以满足您的需求。该函数的逻辑基本上是将字符串 json 记录加载为 json 对象，并将某些键的值转换为字符串。

   class FixJson(beam.DoFn):
       def __init__(self):
           beam.DoFn.__init__(self)
       def process(self, element, *args, **kwargs):
           row = str(element)
           json_obj = json.loads(row)
           json_obj["field"] = str(json_obj["field"])
           return [json_obj]
   
   ...
   
   "Fix json" >> beam.ParDo(FixJson())\
   

3. 最后，在最后一步中，我们将数据保存到 BigQuery。要指定将保存数据的表，我们使用之前创建的 table_spec 变量，如您在代码中所见。

   table_spec = bigquery.TableReference(
       projectId='<your-project>',
       datasetId='<your-dataset>',
       tableId='<your-table>')
   
   ...
   
   "Write to bq" >> beam.io.WriteToBigQuery(table_spec,
                                           custom_gcs_temp_location = '<some-temporary-bucket>',                                            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,                                          create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
   

最后，我想发表两点意见:

1. 此代码假定表已创建并具有正确的架构。

2. 我用下面的示例数据测试了这段代码:

   {"field" : 123}
   {"field" : 23}
   {"field" : 3}
   {"field" : "9123"}
   {"field" : "45"}
   {"field" : "12"}
   {"field" : 1}
   {"field" : "13"}