求分享：使用Python+Cloud Dataflow将MySQL数据导入BigQuery的经验或文章

一、前置准备 #

• 安装依赖包：执行 pip install apache-beam[gcp,mysql] 安装Dataflow所需的GCP和MySQL连接器
• 配置权限：
  • 给Dataflow服务账号分配BigQuery数据写入权限、Cloud Storage读写权限（用于临时文件存储）
  • 确保Dataflow集群能访问MySQL实例（如果是GCP Cloud SQL，可配置VPC私有访问或开放公网白名单）
• 整理连接信息：提前准备MySQL的主机地址、端口、数据库名、用户名、密码，建议用GCP Secret Manager存储敏感凭据，避免硬编码

二、核心代码实现 #

1. 基础Pipeline配置与MySQL读取 #

import apache_beam as beam
from apache_beam.io import ReadFromMySQL
from apache_beam.options.pipeline_options import (
    PipelineOptions, GoogleCloudOptions, StandardOptions
)

def run_mysql_to_bigquery():
    # 初始化Dataflow Pipeline配置
    pipeline_options = PipelineOptions()
    gcp_options = pipeline_options.view_as(GoogleCloudOptions)
    gcp_options.project = "你的GCP项目ID"
    gcp_options.region = "us-central1"  # 根据实际区域调整
    gcp_options.job_name = "mysql-to-bq-sync-job"
    gcp_options.staging_location = "gs://你的存储桶路径/staging"
    gcp_options.temp_location = "gs://你的存储桶路径/temp"
    pipeline_options.view_as(StandardOptions).runner = "DataflowRunner"

    # MySQL连接与读取配置
    mysql_params = {
        "server": "MySQL主机IP或域名",
        "port": 3306,
        "database": "目标数据库名",
        "username": "MySQL用户名",
        "password": "MySQL密码",
        "query": "SELECT id, username, create_time, amount FROM user_info"  # 按需指定查询语句
        # 也可用table_name参数直接指定表："table_name": "user_info"
    }

    with beam.Pipeline(options=pipeline_options) as p:
        # 从MySQL读取数据
        raw_mysql_data = p | "读取MySQL数据" >> ReadFromMySQL(**mysql_params)

2. 数据转换（可选，按需调整） #

如果MySQL与BigQuery的字段名、数据格式不匹配，添加转换逻辑：

def format_for_bigquery(record):
    # 示例：字段重命名、类型转换、格式适配
    return {
        "user_id": record["id"],
        "user_name": record["username"],
        "create_timestamp": record["create_time"].isoformat(),  # 适配BigQuery时间格式
        "total_amount": float(record["amount"])  # 转换数值类型
    }

transformed_data = raw_mysql_data | "转换数据格式" >> beam.Map(format_for_bigquery)

3. 写入BigQuery #

# BigQuery写入配置
bq_params = {
    "table": "你的GCP项目ID:数据集ID.bq目标表名",
    "write_disposition": beam.io.BigQueryDisposition.WRITE_APPEND,  # 可选WRITE_TRUNCATE（覆盖）/WRITE_EMPTY（仅空表写入）
    "create_disposition": beam.io.BigQueryDisposition.CREATE_IF_NEEDED  # 自动建表（需字段类型匹配）
}

# 执行写入
transformed_data | "写入BigQuery" >> beam.io.WriteToBigQuery(**bq_params)

if __name__ == "__main__":
    run_mysql_to_bigquery()

三、实操关键技巧 #

• 增量同步实现：基于MySQL表的自增ID或更新时间戳做增量过滤，例如把查询语句改为 SELECT * FROM user_info WHERE update_time > '2024-01-01 00:00:00'，并将上次同步的时间戳存储在GCS或Cloud Firestore中，每次运行前读取更新
• 性能优化：
  • 针对大表，使用分区查询拆分数据（如按ID范围分批次读取），避免MySQL单查询过载
  • 调整Dataflow的机器类型（如n2-standard-4）和并行度参数，提升处理速度
  • 启用BigQuery批量写入模式，减少API调用频率
• 错误处理：给转换步骤添加异常捕获逻辑，将处理失败的记录写入GCS死信队列，方便后续排查修复：

def safe_transform(record):
    try:
        return format_for_bigquery(record)
    except Exception as e:
        return {"error": str(e), "raw_record": str(record)}

# 拆分正常数据与错误数据
valid_data, error_data = transformed_data | beam.Partition(
    lambda elem, _: 0 if "error" not in elem else 1, 2
)

# 错误数据写入GCS
error_data | "写入死信队列" >> beam.io.WriteToText("gs://你的存储桶/error_logs")

四、场景化替代方案 #

• 若使用GCP Cloud SQL MySQL，可先通过Cloud SQL导出功能将数据转储到GCS，再用BigQuery的批量加载API导入，适合超大规模全量数据迁移
• 如需实时同步，可结合Debezium捕获MySQL的CDC日志，再通过Dataflow消费日志并写入BigQuery

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