如何在GCP Dataflow中实现基于BigQuery的用户数据校验与过滤

对应需求的分步实现 #

步骤1：读取文本文件生成用户ID集合 #

先通过Apache Beam读取文本文件，解析出需要匹配的用户ID并转为集合（提升后续查找效率）：

import apache_beam as beam
import json

def parse_user_data(line):
    # 解析每行的JSON数组，提取用户ID
    users = json.loads(line)
    for user in users:
        yield user['user']

with beam.Pipeline() as p:
    # 生成存储用户ID的PCollection（转为集合）
    user_id_set = (
        p
        | '读取用户数据文件' >> beam.io.ReadFromText('gs://your-bucket/user_data.txt')
        | '解析用户ID' >> beam.FlatMap(parse_user_data)
        | '转为用户集合' >> beam.transforms.combiners.ToSet()
    )

步骤2：读取BigQuery表并过滤不匹配行 #

读取目标BigQuery表，将用户ID集合作为侧输入，过滤出不在集合中的行：

def filter_unmatched(row, user_set):
    # 校验当前行的user是否不在用户集合中，是则返回该行
    if row['user'] not in user_set:
        return row

# 读取BigQuery全量数据
bq_source_rows = (
    p
    | '读取BigQuery表' >> beam.io.ReadFromBigQuery(
        query='SELECT * FROM `your-project.your_dataset.target_table`',
        use_standard_sql=True
    )
)

# 输出不匹配的行
unmatched_rows = (
    bq_source_rows
    | '过滤不匹配记录' >> beam.FlatMap(
        filter_unmatched,
        user_set=beam.pvalue.AsSingleton(user_id_set)
    )
)

步骤3：输出结果 #

将过滤得到的不匹配行写入目标存储（示例为BigQuery）：

unmatched_rows | '写入不匹配结果' >> beam.io.WriteToBigQuery(
    table='your-project.your_dataset.unmatched_results',
    schema='user:STRING, age:INTEGER, ...',  # 匹配原表字段结构
    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
)

验证PCollection数据是否存在于BigQuery lookup表的方法 #

方法1：侧输入集合匹配（小数据量场景） #

从BigQuery lookup表提取所有用户ID转为集合，作为侧输入校验PCollection中的数据：

def check_existence(user_id, bq_user_set):
    return {'user': user_id, 'exists_in_bq': user_id in bq_user_set}

# 读取BigQuery lookup表的用户ID并转为集合
bq_user_id_set = (
    p
    | '读取Lookup表用户ID' >> beam.io.ReadFromBigQuery(
        query='SELECT user FROM `your-project.your_dataset.lookup_table`',
        use_standard_sql=True
    )
    | '提取用户ID' >> beam.Map(lambda row: row['user'])
    | '转为Lookup集合' >> beam.transforms.combiners.ToSet()
)

# 执行存在性校验
validation_results = (
    user_id_set  # 这里的user_id_set是之前从文本生成的PCollection
    | '校验存在性' >> beam.Map(
        check_existence,
        bq_user_set=beam.pvalue.AsSingleton(bq_user_id_set)
    )
)

方法2：BigQuery JOIN查询（大数据量场景） #

当PCollection数据量较大时，用侧输入集合可能导致内存不足，可通过临时表+JOIN的方式批量验证：

# 将PCollection数据写入BigQuery临时表
user_id_set | '写入临时用户表' >> beam.io.WriteToBigQuery(
    table='your-project.your_dataset.temp_user_ids',
    schema='user:STRING',
    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
)

# 用JOIN查询验证存在性
validation_query = """
SELECT t.user, CASE WHEN l.user IS NOT NULL THEN TRUE ELSE FALSE END AS exists_in_bq
FROM `your-project.your_dataset.temp_user_ids` t
LEFT JOIN `your-project.your_dataset.lookup_table` l
ON t.user = l.user
"""

validation_results = (
    p
    | '执行校验查询' >> beam.io.ReadFromBigQuery(
        query=validation_query,
        use_standard_sql=True
    )
)

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