Apache Beam流式Pipeline无报错但未写入BigQuery求助

我正在构建一个运行于GCP、以DataFlow为执行器的Apache Beam数据流水线。代码执行无报错，但BigQuery未写入数据；查看作业执行图发现“写入BigQuery”环节无活动，不过PubSub数据输入环节有数据流入。我已添加触发器，但仍无数据输出，恳请协助排查问题。

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.io.gcp.bigquery import WriteToBigQuery
from apache_beam.dataframe.convert import to_dataframe, to_pcollection
from apache_beam import window
import typing
import numpy as np
import pandas as pd


class BmsSchema(typing.NamedTuple):
  can_data_frame_1: typing.Optional[str]


beam.coders.registry.register_coder(BmsSchema, beam.coders.RowCoder)


class ParsePubSubMessage(beam.DoFn):
  def process(self, message):
      import json
      all_columns = [
          "can_data_frame_1"
      ]
      main_dict = dict(zip(all_columns, [None] * len(all_columns)))
      record = json.loads(message.decode('utf-8'))
      main_dict.update(record)
      yield {
          all_columns[0]: main_dict[all_columns[0]]}


def run():
  options = PipelineOptions(
      project='dwingestion',
      runner='DataflowRunner',
      streaming=True,
      temp_location='gs://....../temp',
      staging_location='gs://.........../staging',
      region='europe-west1',
      job_name='.........streaming-pipeline-dataflow',
      save_main_session=True,
      flags=['--allow_unsafe_triggers']
  )

  options.view_as(StandardOptions).streaming = True

  input_subscription = 'projects/..._data_streaming'

  table_schema = {
      "fields": [
          {"name": "current_mA", "type": "INTEGER", "mode": "NULLABLE"}
      ]
  }

  with beam.Pipeline(options=options) as p:
      messages = (p
                  | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription=input_subscription)
                  | 'Apply Fixed Window' >> beam.WindowInto(
                      window.FixedWindows(60),
                      trigger=beam.trigger.AfterWatermark(),
                      allowed_lateness=window.Duration(10),
                      accumulation_mode=beam.trigger.AccumulationMode.DISCARDING
                  )
                  | 'Parse PubSub Message' >> beam.ParDo(ParsePubSubMessage())
                  | 'Attaching the schema' >> beam.Map(lambda x: BmsSchema(**x)).with_output_types(BmsSchema)
                  )

      # Convert the messages to a DataFrame
      df = to_dataframe(messages)
      

      # Extract and process the 'current_mA' field
      df['current_mA'] = df['can_data_frame_1'].str[4:8].apply(lambda x: int(x, 16) if pd.notna(x) else 0)
      df['current_mA'] = df['current_mA'].where(df['current_mA'] < 0x8000, df['current_mA'] - 0x10000)
      df['current_mA'] = df['current_mA'] * 10

      
      # Convert back to PCollection and map to dictionaries
      transformed_pcol = (
          to_pcollection(df)
          | 'Log Transformed PCollection' >> beam.Map(lambda x: (print(f"Transformed Row: {x}"), x)[1])  # Debugging
          | 'Convert to Dict with Native Types' >> beam.Map(lambda row: {
              "current_mA": int(row.current_mA) if row.current_mA is not None else None
          })
      )

      # Write to BigQuery
      transformed_pcol | 'Write to BigQuery' >> WriteToBigQuery(
          table='..........table_test_all_columns_04',
          write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
          create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
          schema=table_schema,
          custom_gcs_temp_location='gs://......_template/temp'
      )


if __name__ == '__main__':
  run()

1. Beam DataFrame流处理触发问题 #

流式场景下，to_dataframe转换后的DataFrame操作默认依赖全局窗口的默认触发机制，你之前设置的FixedWindow触发器不会传递到DataFlow处理环节。这会导致DataFrame一直等待数据积累，不会主动触发输出，后续的to_pcollection自然没有数据流入。

解决办法：

• 优先改用原生PCollection操作替代DataFrame处理，避免流式场景下的触发兼容问题；
• 若必须使用DataFrame，需显式为DataFrame操作配置触发策略，或在转换回PCollection前添加窗口触发逻辑。

2. 窗口触发器配置问题 #

当前使用的AfterWatermark()触发器需要等待窗口的水印推进到窗口结束时间才会输出数据。如果PubSub消息未携带正确的事件时间戳，或时间戳设置为远未来值，水印无法正常推进，窗口永远不会触发，数据会一直卡在窗口中。

解决办法：

• 检查PubSub消息是否包含合法的事件时间戳，或在读取时显式指定时间戳属性：

  | 'Read from PubSub with Timestamp' >> beam.io.ReadFromPubSub(subscription=input_subscription).with_timestamp_attribute('event_timestamp')
  

• 临时替换为基于处理时间的触发器验证，比如AfterProcessingTime(60)（处理时间后60秒触发）：

  | 'Apply Fixed Window' >> beam.WindowInto(
      window.FixedWindows(60),
      trigger=beam.trigger.AfterProcessingTime(60),
      allowed_lateness=window.Duration(10),
      accumulation_mode=beam.trigger.AccumulationMode.DISCARDING
  )
  

3. 数据转换环节潜在错误 #

• 检查ParsePubSubMessage输出的字典是否正确包含can_data_frame_1字段，若PubSub消息中无该字段，后续DataFrame处理会生成空值，可能导致数据被过滤；
• 查看DataFlow Worker日志，确认Log Transformed PCollection步骤是否有打印日志，若无则说明DataFrame转换后无数据输出；
• 验证current_mA计算逻辑：比如can_data_frame_1字符串长度是否足够4-8位，若长度不足，str[4:8]返回空字符串，int(x,16)会抛出异常导致数据丢失。

4. BigQuery写入配置验证 #

• 确认custom_gcs_temp_location指定的GCS路径存在，且DataFlow服务账号拥有读写权限；
• 检查BigQuery表schema与输出字典是否匹配（当前配置为NULLABLE，空值不会导致写入失败，但需确认数据类型是否一致）；
• 查看DataFlow作业的BigQuery写入监控指标，排查是否有写入失败记录。

快速验证步骤 #

1. 临时移除DataFrame处理逻辑，直接将解析后的PCollection转换为符合BigQuery schema的字典并写入，验证基础流程是否正常；
2. 在窗口操作后添加beam.Map(print)，查看窗口是否有数据输出；
3. 改用DirectRunner本地运行，通过控制台日志追踪数据流动路径，定位阻塞点。

内容的提问来源于stack exchange，提问作者oyugi.collins