Apache Beam on Google DataFlow：从主方法中收集指标

问题根源 #

你遇到的UnsupportedException是因为DataFlow Runner（云端分布式执行）不支持通过本地PipelineResult对象直接获取作业指标。Beam的指标API在DirectRunner（本地运行）下可以正常获取，但DataFlow作业在云端集群执行时，本地进程无法直接访问远程作业的指标数据，而Cloud Console显示的指标是DataFlow服务独立收集并展示的，和本地API不是同一通路。

可行解决办法 #

1. 通过DataFlow API/CLI拉取作业指标

作业执行完成后，你可以通过Google Cloud的官方工具查询已完成作业的指标：

• gcloud命令行方式：
  先获取作业ID（可从Cloud Console或gcloud dataflow jobs list拿到），然后执行：

  gcloud dataflow metrics list <JOB_ID> --region=<REGION>
  

  可通过--filter参数筛选目标统计项，比如：

  gcloud dataflow metrics list <JOB_ID> --region=us-central1 --filter="name=total_records OR name=null_column_count"
  

• 编程方式（以Python为例）：
  使用Google Cloud的DataFlow客户端库查询：

  from google.cloud import dataflow_v1beta3
  
  client = dataflow_v1beta3.JobsV1Beta3Client()
  job_path = client.job_path("<PROJECT_ID>", "<REGION>", "<JOB_ID>")
  metrics = client.get_job_metrics(job_path)
  
  # 提取目标统计值
  for metric in metrics.metrics:
      if metric.name.name in ["total_records", "null_column_count"]:
          print(f"{metric.name.name}: {metric.scalar_value.value}")
  

2. 将统计结果写入外部存储（推荐批处理场景）

直接在Beam管道中添加统计分支，将计算结果写入BigQuery、Cloud Storage或Cloud Firestore，作业完成后直接查询存储即可获取统计数据：

• 示例思路（以Python为例）：

  import apache_beam as beam
  from apache_beam.transforms.combiners import Count
  
  def count_nulls(element):
      return 1 if element.get("target_column") is None else 0
  
  with beam.Pipeline(options=pipeline_options) as p:
      # 主管道：Cassandra到Kafka
      cassandra_data = p | "Read from Cassandra" >> beam.io.ReadFromCassandra(...)
      cassandra_data | "Write to Kafka" >> beam.io.WriteToKafka(...)
  
      # 统计总记录数并写入BigQuery
      total_records = cassandra_data | "Count total" >> Count.Globally()
      total_records | "Write total to BQ" >> beam.io.WriteToBigQuery(
          table="<PROJECT>:<DATASET>.stats_table",
          schema="metric:STRING, value:INTEGER",
          write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
      )
  
      # 统计空值记录数并写入BigQuery
      null_count = cassandra_data | "Map nulls" >> beam.Map(count_nulls) | "Sum nulls" >> beam.CombineGlobally(sum)
      null_count | "Write null count to BQ" >> beam.io.WriteToBigQuery(
          table="<PROJECT>:<DATASET>.stats_table",
          schema="metric:STRING, value:INTEGER",
          write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
      )
  

  这种方式更适配批处理场景，统计结果与作业执行绑定，无需额外查询API，且结果持久化可追溯。

3. 注意事项

• 打包Flex Template时，确保包含所需依赖（如Google Cloud客户端库、BigQuery连接器等）。
• 作业使用的服务账号需拥有对应权限：比如访问DataFlow API的dataflow.jobs.getMetrics权限，或写入目标存储的权限。
• 自定义指标命名要清晰，避免与系统指标冲突，方便后续筛选查询。

内容的提问来源于stack exchange，提问作者sunitha