如何用Apache Beam实现持续读取BigQuery的流式数据管道？

核心思路 #

BigQuery非流式插入的基表不支持原生流式读取，因此我们采用流管道+周期性触发读取最新表快照的方案，替代定时批处理全量扫描，同时搭配全局窗口触发器实现每12小时输出唯一用户总数。

具体实现步骤 #

1. 构建持续运行的流管道触发源：用定时生成的信号维持管道运行，避免单次执行后停止
2. 周期性读取BigQuery数据：每次触发时读取基表最新数据，全量/增量读取可根据表更新模式选择
3. 窗口与触发器配置：用全局窗口+重复触发器，每12小时触发一次唯一用户数计算

Python代码示例 #

全量更新基表场景（每12小时读取全表计算当前总数）

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms.window import GlobalWindows, AfterProcessingTime, Repeatedly

def run_pipeline():
    # 配置管道为流模式
    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        # 每12小时生成触发信号（单位：秒）
        trigger_signals = p | "生成周期性触发信号" >> beam.GenerateSequence(
            start=0,
            interval=12 * 3600,
            stop=None
        )

        # 每次触发读取BigQuery最新基表数据，禁用缓存确保读取最新内容
        user_records = trigger_signals | "读取BigQuery基表" >> beam.FlatMap(
            lambda _: beam.io.ReadFromBigQuery(
                query='SELECT user_id FROM `你的项目ID.数据集ID.用户基表`',
                use_standard_sql=True,
                use_query_cache=False
            )
        )

        # 全局窗口+每12小时重复触发，DISCARDING模式清空历史累积，计算当前全量唯一用户数
        unique_user_count = user_records | "窗口配置" >> beam.WindowInto(
            GlobalWindows(),
            trigger=Repeatedly(AfterProcessingTime(12 * 3600)),
            accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING
        ) | "去重计数" >> beam.CombineGlobally(
            lambda elements: len(set(element['user_id'] for element in elements))
        )

        # 输出结果（可替换为写入BigQuery/Cloud Storage等目标）
        unique_user_count | "输出结果" >> beam.Map(
            lambda count: f"[{beam.transforms.util.timestamp_to_rfc3339(beam.transforms.util.current_timestamp())}] 唯一用户总数: {count}"
        ) | beam.io.WriteToText(
            './unique_user_counts',
            append=True
        )

if __name__ == '__main__':
    run_pipeline()

增量更新基表场景（仅读取上次触发后的新增数据，优化扫描效率）

如果基表含update_time等时间字段记录更新信息，可通过状态管理维护上次触发时间，只读取增量数据：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from apache_beam.transforms.window import GlobalWindows, AfterProcessingTime, Repeatedly
import time
from apache_beam.coders import TimestampCoder

class TrackLastTriggerTime(beam.DoFn):
    # 维护上次触发时间的状态
    LAST_TRIGGER = beam.DoFn.StateParam(beam.StateSpec('last_trigger', TimestampCoder()))
    # 定时触发计时器
    TRIGGER_TIMER = beam.DoFn.TimerParam(beam.TimerSpec('trigger_timer', time_domain=beam.TimeDomain.REAL_TIME))

    def process(self, _, last_trigger=LAST_TRIGGER, timer=TRIGGER_TIMER):
        current_ts = time.time()
        last_ts = last_trigger.read()
        
        yield {'last_trigger_time': last_ts, 'current_time': current_ts}
        
        # 更新上次触发时间
        last_trigger.write(current_ts)
        # 设置下一次触发时间
        timer.set(current_ts + 12 * 3600)

def run_incremental_pipeline():
    options = PipelineOptions()
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        # 初始触发信号，启动后立即执行第一次读取
        initial_trigger = p | "初始触发" >> beam.Create([0])

        # 维护触发时间并生成定时触发事件
        trigger_events = initial_trigger | "跟踪触发时间" >> beam.ParDo(TrackLastTriggerTime())

        # 根据上次触发时间读取增量数据，首次运行读取全表
        user_records = trigger_events | "读取增量数据" >> beam.FlatMap(
            lambda event: beam.io.ReadFromBigQuery(
                query=f"""
                    SELECT user_id 
                    FROM `你的项目ID.数据集ID.用户基表`
                    WHERE update_time >= TIMESTAMP_SECONDS({event['last_trigger_time']})
                    """ if event['last_trigger_time'] is not None else "SELECT user_id FROM `你的项目ID.数据集ID.用户基表`",
                use_standard_sql=True
            )
        )

        # 全局窗口累积计算总唯一用户数，ACCUMULATING模式保留历史数据
        total_unique_users = user_records | "窗口配置" >> beam.WindowInto(
            GlobalWindows(),
            trigger=Repeatedly(AfterProcessingTime(12 * 3600)),
            accumulation_mode=beam.transforms.trigger.AccumulationMode.ACCUMULATING
        ) | "去重累积计数" >> beam.CombineGlobally(
            lambda elements: len(set(element['user_id'] for element in elements))
        )

        # 输出结果
        total_unique_users | "输出结果" >> beam.Map(
            lambda count: f"[{beam.transforms.util.timestamp_to_rfc3339(beam.transforms.util.current_timestamp())}] 累计唯一用户总数: {count}"
        ) | beam.io.WriteToText(
            './total_unique_users',
            append=True
        )

if __name__ == '__main__':
    run_incremental_pipeline()

关键注意事项 #

• 禁用查询缓存：设置use_query_cache=False，避免读取到旧缓存数据
• 累积模式选择：DISCARDING输出当前全量用户数，ACCUMULATING输出累计历史用户数
• 分区表优化：如果基表按时间分区，利用_PARTITIONTIME字段过滤，进一步减少扫描量

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