使用Dataflow从Cloud SQL MySQL向BigQuery写入数据失败排查

核心问题：数据读取逻辑错误 #

你的ReadFromJdbc类中存在致命逻辑错误，导致没有任何数据被输出到后续Pipeline环节：
原代码中，while循环不断调用fetchmany(1000)覆盖rows变量，直到最后一次返回空结果才跳出循环，此时rows是空值，后续的for row in rows循环根本不会执行，自然没有数据被yield出来。

修复后的ReadFromJdbc代码 #

把数据处理逻辑放到while循环内部，每次获取到数据就立即处理并输出：

class ReadFromJdbc(beam.DoFn):
    def process(self, context):
        # Connection setup
        connection = jaydebeapi.connect(
            'com.mysql.cj.jdbc.Driver',
            jdbc_url,
            [username, password],
            '/temp/mysql-connector-j-8.2.0.jar',
        )
        
        cursor = connection.cursor()
        cursor.execute(sql) # 'select col1,col2 from mysql_table';

        while True:
            rows = cursor.fetchmany(1000)  
            if not rows:
                break
            # 每次获取到数据就立即处理并yield
            for row in rows:
                processed_row = tuple(int(value) if isinstance(value, java.lang.Integer) else value for value in row)
                yield processed_row

        cursor.close()
        connection.close()

两种数据格式的BigQuery写入适配 #

1. MySQL JSON_OBJECT格式输出 #

如果使用SELECT JSON_OBJECT('col1', col1,'col2', col2) AS payload FROM employee，返回的是单个字符串字段的元组，需要调整Pipeline适配：

• 方式一：修改BigQuery schema为payload:STRING直接写入
• 方式二：添加JSON解析步骤，转换为字典后匹配原schema：

def parse_json_payload(element):
    import json
    return json.loads(element[0])

mysql_data = (
    data     
    | 'Read From Jdbc' >> beam.ParDo(ReadFromJdbc())
    | 'Parse JSON' >> beam.Map(parse_json_payload)
    | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
        table_spec,
        schema='col1:INTEGER,col2:STRING',
        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)
)

2. 字典格式输出 #

如果用dict(zip(row_headers, rv))生成字典，需要在ReadFromJdbc中正确获取表头并返回字典：

class ReadFromJdbc(beam.DoFn):
    def process(self, context):
        connection = jaydebeapi.connect(
            'com.mysql.cj.jdbc.Driver',
            jdbc_url,
            [username, password],
            '/temp/mysql-connector-j-8.2.0.jar',
        )
        
        cursor = connection.cursor()
        cursor.execute(sql)
        # 获取查询结果的表头
        row_headers = [x[0] for x in cursor.description]
        
        while True:
            rows = cursor.fetchmany(1000)  
            if not rows:
                break
            for rv in rows:
                # 转换为字典格式
                row_dict = dict(zip(row_headers, rv))
                # 按需做类型转换
                row_dict['col1'] = int(row_dict['col1'])
                yield row_dict

        cursor.close()
        connection.close()

这种格式可以直接匹配BigQuery的schema，无需额外解析步骤。

额外注意事项 #

• 网络访问：确保Dataflow集群能访问Cloud SQL实例，公共IP需授权Dataflow服务账号权限，私有IP需配置VPC连接器
• 依赖管理：JDBC驱动包需在Dataflow集群中可用，建议通过setup.py打包依赖或使用GCS存储的驱动路径
• 日志排查：即使Pipeline显示成功，也需查看Dataflow日志面板，排查是否存在隐藏的类型转换、权限等异常

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