数据流作业-写入BigQuery时出错

Question

使用“file_loads”技术通过Apache Beam数据流作业写入BigQuery时出错。流式插入（else块）工作正常，符合预期。file_load（如果块）失败，错误在代码后面给出。bucket中GCS上的临时文件是有效的JSON对象。

来自pub/sub的原始事件示例：

"{'event': 'test', 'entityId': 13615316690, 'eventTime': '2020-08-12T15:56:07.130899+00:00', 'targetEntityId': 8947793, 'targetEntityType': 'item', 'entityType': 'guest', 'properties': {}}" 
 
"{'event': 'test', 'entityId': 13615316690, 'eventTime': '2020-08-12T15:56:07.130899+00:00', 'targetEntityId': 8947793, 'targetEntityType': 'item', 'entityType': 'guest', 'properties': {‘action’: ‘delete’}}"  

from __future__ import absolute_import

import logging
import sys
import traceback
import argparse
import ast
import json
import datetime
import dateutil.parser as date_parser

import apache_beam as beam
import apache_beam.pvalue as pvalue
from google.cloud.bigquery import CreateDisposition, WriteDisposition
from apache_beam.io.gcp.bigquery_tools import RetryStrategy

def get_values(element):
    # convert properties from dict to arr of dicts to form a repeatable bq table record
    prop_list = [{'property_name': k, 'property_value': v} for k, v in element['properties'].items()]
    date_parsed = date_parser.parse(element.get('eventTime'))
    event_time = date_parsed.strftime('%Y-%m-%d %H:%M:00')
    
    raw_value = {'event': element.get('event'),
                 'entity_type': element.get('entityType'),
                 'entity_id': element.get('entityId'),
                 'target_entity_type': element.get('targetEntityType'),
                 'target_entity_id': element.get('targetEntityId'),
                 'event_time': event_time,
                 'properties': prop_list
                 }

    return raw_value

def stream_to_bq(c: dict):
    argv = [
        f'--project={c["PROJECT"]}',
        f'--runner=DataflowRunner',
        f'--job_name={c["JOBNAME"]}',
        f'--save_main_session',
        f'--staging_location=gs://{c["BUCKET_NAME"]}/{c["STAGING_LOCATION"]}',
        f'--temp_location=gs://{c["BUCKET_NAME"]}/{c["TEMP_LOCATION"]}',
        f'--network={c["NETWORKPATH"]}',
        f'--subnetwork={c["SUBNETWORKPATH"]}',
        f'--region={c["REGION"]}',
        f'--service_account_email={c["SERVICE_ACCOUNT"]}',
        # f'--setup_file=./setup.py',
        # f'--autoscaling_algorithm=THROUGHPUT_BASED',
        # f'--maxWorkers=15',
        # f'--experiments=shuffle_mode=service',
        '--no_use_public_ips',
        f'--streaming'
    ]

    if c['FILE_LOAD']:
        argv.append('--experiments=allow_non_updatable_job')
        argv.append('--experiments=use_beam_bq_sink')

    p = beam.Pipeline(argv=argv)
    valid_msgs = (p
                          | 'Read from Pubsub' >>
                          beam.io.ReadFromPubSub(subscription=c['SUBSCRIPTION']).with_output_types(bytes)
                          )

    records = (valid_msgs
               | 'Event Parser(BQ Row) ' >> beam.Map(get_values)
               )

    # Load data to BigQuery using - 'Load Jobs' or 'Streaming Insert', choice based on latency expectation.
    if c['FILE_LOAD']:
        records | 'Write Result to BQ' >> beam.io.WriteToBigQuery(c["RAW_TABLE"],
                                                                  project=c["PROJECT"],
                                                                  dataset=c["DATASET_NAME"],
                                                                  method='FILE_LOADS',
                                                                  triggering_frequency=c['FILE_LOAD_FREQUENCY'],
                                                                  create_disposition=CreateDisposition.CREATE_NEVER,
                                                                  write_disposition=WriteDisposition.WRITE_APPEND
                                                                  )

        
    else:
        records | 'Write Result to BQ' >> beam.io.WriteToBigQuery(c["RAW_TABLE"],
                                                                  project=c["PROJECT"],
                                                                  dataset=c["DATASET_NAME"],
                                                                  create_disposition=CreateDisposition.CREATE_NEVER,
                                                                  write_disposition=WriteDisposition.WRITE_APPEND,
                                                                  insert_retry_strategy=RetryStrategy.RETRY_ON_TRANSIENT_ERROR
                                                                  )

    

    p.run()

数据流作业出错：

message: 'Error while reading data, error message: JSON table encountered too many errors, giving up. Rows: 1; errors: 1. Please look into the errors[] collection for more details.' reason: 'invalid'> [while running 'generatedPtransform-1801'] java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895) org.apache.beam.sdk.util.MoreFutures.get(MoreFutures.java:57)

Answer 1

这个问题看起来像是错误地加载到了BigQuery。我的建议是尝试在数据流之外进行测试加载作业，以确保您的模式和数据结构良好。您可以遵循以下BQ文档。

另外，我注意到您没有指定模式或schema_autodetect。我建议你一定要指定。

为了理解错误，尝试检查数据流作业日志，其中可能包含大量信息。如果您的加载作业失败了，您可以在BigQuery中检查这些作业，它们还会为您提供更多关于失败的信息。您可以使用此StackDriver日志查找BQ加载作业ID：

resource.type="dataflow_step"
resource.labels.job_id= < YOUR DF JOB ID >
jsonPayload.message:("Triggering job" OR "beam_load")

        schema = {
            "fields":
                [
                    {
                        "name": "name",
                        "type": "STRING"
                    },
                    {
                        "name": "repeated",
                        "type": "RECORD",
                        "mode": "REPEATED",
                        "fields": [
                            {
                                "name": "spent",
                                "type": "INTEGER"
                            },
                            {
                                "name": "ts",
                                "type": "TIMESTAMP"
                            }
                        ]
                    }
                ]
            }

        def fake_parsing(element):
            # Using a fake parse so it's easier to reproduce
            properties = []

            rnd = random.random()
            if rnd < 0.25:
                dict_prop = {"spent": random.randint(0, 100),
                             "ts": datetime.now().strftime('%Y-%m-%d %H:%M:00')}
                properties.append(dict_prop)
            elif rnd > 0.75:
                # repeated
                dict_prop = {"spent": random.randint(0, 100),
                             "ts": datetime.now().strftime('%Y-%m-%d %H:%M:00')}
                properties += [dict_prop, dict_prop]
            elif 0.5 > rnd > 0.75:
                properties.append({"ts": datetime.now().strftime('%Y-%m-%d %H:%M:00')})

            return {"name": 'inigo',
                    "repeated": properties}

        pubsub = (p | "Read Topic" >> ReadFromPubSub(topic=known_args.topic)
                    | "To Dict" >> beam.Map(fake_parsing))

        pubsub | "Stream To BQ" >> WriteToBigQuery(
            table=f"{known_args.table}_streaming_insert",
            schema=schema,
            write_disposition=BigQueryDisposition.WRITE_APPEND,
            method="STREAMING_INSERTS")

        pubsub | "Load To BQ" >> WriteToBigQuery(
            table=f"{known_args.table}_load_job",
            schema=schema,
            write_disposition=BigQueryDisposition.WRITE_APPEND,
            method=WriteToBigQuery.Method.FILE_LOADS,
            triggering_frequency=known_args.triggering,
            insert_retry_strategy="RETRY_ON_TRANSIENT_ERROR")