带有混合工作负载的数据流作业-流式插入和加载作业（Python）

Question

https://cloud.google.com/blog/products/data-analytics/how-to-how高效处理实时和聚合数据

用例说明步骤：

1. 从pubsub获取流式原始事件。
2. 验证接收的原始事件。
3. 筛选特定类型的事件。
4. 创建筛选事件的字典。
5. 同时，将筛选的事件通过窗口操作传递并聚合。
6. 2种输出类型-原始事件字典、聚合事件字典。
7. 按照上面链接中解释的设计，原始事件字典属于低紧急度类别，聚合事件属于高紧急度类别。
8. 为原始事件尝试“file_loads”方法以避免成本部分。
9. 正在为聚合事件尝试'streaming_insert'方法，因为该方法需要实时可用。

p = beam.Pipeline(argv=argv)
    valid_msgs, errors = (p
                          | 'Read from Pubsub' >>
                          beam.io.ReadFromPubSub(subscription=c['SUBSCRIPTION']).with_output_types(bytes)
                          | 'Validate PubSub Event' >> beam.ParDo(ValidateMessages()).with_outputs('errors', main='valid')
                          )

    filtered_events = (valid_msgs | 'Filter Events' >> beam.Filter(filter_msgs))

    raw_events = (filtered_events | 'Prepare Raw Event Row for BQ ' >> beam.Map(get_raw_values))

    agg_events = (filtered_events
                  | f'Streaming Window for {c["WINDOW_TIME"]} seconds' >> beam.WindowInto(window.FixedWindows(c['WINDOW_TIME']))
                  | 'Event Parser' >> beam.Map(get_agg_values)
                  | 'Event Aggregation' >> beam.CombinePerKey(sum)
                  | 'Prepare Aggregate Event Row for BQ' >> beam.Map(get_count)
                  )

    # Raw events are written to BigQuery using 'Load Jobs' every 10 minutes.
    write_result_raw = (raw_events | 'Write Raw Events to BQ' >> beam.io.WriteToBigQuery(c["RAW_TABLE"],
                                                                                         project=c["PROJECT"],
                                                                                         dataset=c["DATASET_NAME"],
                                                                                         method='FILE_LOADS',
                                                                                         triggering_frequency=10))

    # Aggregated events are written to BigQuery using 'Streaming Inserts'.
    write_result_agg = (agg_events | 'Write Aggregate Results to BQ' >> beam.io.WriteToBigQuery(c["COUNT_TABLE"],
                                                                                                project=c["PROJECT"],
                                                                                                dataset=c["DATASET_NAME"],
                                                                                                create_disposition=CreateDisposition.CREATE_NEVER,
                                                                                                write_disposition=WriteDisposition.WRITE_APPEND,
                                                                                                insert_retry_strategy=RetryStrategy.RETRY_ALWAYS))

错误：

File "/usr/local/lib/python3.6/site-packages/apache_beam/io/gcp/bigquery.py", line 1493, in expand
42    'triggering_frequency can only be used with '
43ValueError: triggering_frequency can only be used with FILE_LOADS method of writing to BigQuery.

根据@Iñigo响应，我添加了标志。但这也不起作用。请看下面的详细信息。

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

 p = beam.Pipeline(argv=argv)
 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,
                                                                  insert_retry_strategy=RetryStrategy.RETRY_ON_TRANSIENT_ERROR
                                                                  )```

Error from dataflow job.
```Workflow failed.Causes: Because of the shape of your pipeline, the Cloud Dataflow job optimizer produced a job graph that is not updatable using the - -update pipeline option.This is a known issue that we are working to resolve.See https: // issuetracker.google.com / issues / 118375066 for information about how to modify the shape of your pipeline to avoid this error.You can override this error and force the submission of the job by specifying the --experiments=allow_non_updatable_job parameter., The stateful transform named 'Write Errors to BQ/BigQueryBatchFileLoads/ImpulseSingleElementPC/Map(decode).out/FromValue/ReadStream' is in two or more computations.```


EDITS : 08/11/2020

Added both the flags mentioned as the Pipeline argument.

```INFO:root:Argument to Beam Pipeline:['--project=xxxxxx, '--runner=DataflowRunner', '--job_name=df-pubsub-raw', '--save_main_session', '--staging_location=gs:/staging/', '--temp_location=gs://temp/', '--network=dataflow-localnet', '--subnetwork=regions/us-central1/subnetworks/us-central1', '--region=us-central1', '--service_account_email=[e-mail]', '--no_use_public_ips', '--streaming', '--experiments=[allow_non_updatable_job, use_beam_bq_sink]']

INFO:root:File load enabled 
INFO:root:Write using file load with frequency:5  

26  File "./dataflow_ps_stream_bq.py", line 133, in stream_to_bq  27    write_disposition=WriteDisposition.WRITE_APPEND  28  File "/usr/local/lib/python3.6/site-packages/apache_beam/pvalue.py", line 141, in __or__  29    return self.pipeline.apply(ptransform, self)  30  File "/usr/local/lib/python3.6/site-packages/apache_beam/pipeline.py", line 610, in apply  31    transform.transform, pvalueish, label or transform.label)  32  File "/usr/local/lib/python3.6/site-packages/apache_beam/pipeline.py", line 620, in apply  33    return self.apply(transform, pvalueish)  34  File "/usr/local/lib/python3.6/site-packages/apache_beam/pipeline.py", line 663, in apply  35    pvalueish_result = self.runner.apply(transform, pvalueish, self._options)  36  File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/dataflow/dataflow_runner.py", line 153, in apply  37    return super(DataflowRunner, self).apply(transform, input, options)  38  File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/runner.py", line 198, in apply  39    return m(transform, input, options)  40  File "/usr/local/lib/python3.6/site-packages/apache_beam/runners/runner.py", line 228, in apply_PTransform  41    return transform.expand(input)  42  File "/usr/local/lib/python3.6/site-packages/apache_beam/io/gcp/bigquery.py", line 1493, in expand  43    'triggering_frequency can only be used with '  44ValueError: triggering_frequency can only be used with FILE_LOADS method of writing to BigQuery.  ```

Answer 1

您需要添加标志--实验use_beam_bq_sink。这个问题已经存在一段时间了，Dataflow覆盖了load类型。

你可以在光束回购中看到这一点。

它也看起来有一个正在进行的公关来改进插入，包括这一点（我只是略过代码）。