下面是这种情况:我在GCS中有一组压缩文件,并且有一个.gz文件扩展名(即000000_ [0-5] .gz),我试图将其导入到一个BQ表中。到目前为止,我一直在从命令行执行命令,但希望使用Dataflow完成此操作,可能会在将来添加一些转换。
压缩GCS文件中的数据是一个复杂的JSON结构,经常更改模式,因此最简单的方法是将整个文件作为TSV只有一列(称为record
)引入BigQuery,然后在BQ中使用JSON_EXTRACT函数来解析需要时所需的价值。
问题:我编写了一个Dataflow管道,在这种情况下它将做到最低限度;从GCS读取并写入BigQuery表。但是,当我执行此管道时,我收到一个JSON解析错误,如下所示:
Error while reading data, error message: JSON table encountered too
many errors, giving up. Rows: 1; errors: 1., error: Error while reading
data, error message: JSON table encountered too many errors, giving up.
Rows: 1; errors: 1., error: Error while reading data, error message:
JSON parsing error in row starting at position 2630029539: Value
encountered without start of object.
下面是我的Dataflow脚本,其中一些变量是匿名的。
from __future__ import absolute_import
import argparse
import logging
import re
import json
import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.io import Read
from apache_beam.io import WriteToText
from apache_beam.io import WriteToBigQuery
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
def run(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument('--input',
dest='input',
default='gs://BUCKET_NAME/input-data/000000_0.gz',
help='Input file to process.')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_args.extend([
'--runner=DataflowRunner',
'--project=GCP_PROJECT_NAME',
'--staging_location=gs://BUCKET_NAME/dataflow-staging',
'--temp_location=gs://BUCKET_NAME/dataflow-temp',
'--job_name=gcs-gzcomp-to-bq1',
])
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = True
with beam.Pipeline(options=pipeline_options) as p:
(p | "ReadFromGCS" >> ReadFromText(known_args.input)
| WriteToBigQuery('TABLE_NAME', dataset='DATASET_NAME',
project='GCP_PROJECT_NAME', schema='record:string'))
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()
正如您所看到的,我尝试执行与传统加载作业相同的操作,通过指定仅包含一个具有字符串类型的列的模式,但它仍然失败。
有没有办法明确告诉Dataflow有关我如何导入GCS文件的更多详细信息?即指定TSV,即使它是每行上的有效JSON对象?
此外,如果此错误与我可能搞砸的其他任何事情有关,请同时打电话给我;我是Dataflow的新手,但对BQ和其他一些GCP工具很有经验,所以希望将它添加到我的工具箱中。
我相信WriteToBigQuery
的输入集合应该是字典集合(每个键映射到BigQuery列),而不是字符串集合。尝试通过像| beam.Map(lambda line: dict(record=line))
这样的东西。