Vertex AI中Apache Beam从Datastore到BigQuery的JSON加载错误排查

问题根源 #

Datastore的BYTES类型直接解码为字符串后，可能包含不可打印字符或不符合BigQuery JSON解析规范的内容，导致批量加载时触发JSON格式错误。JSON本身不支持原始二进制数据，直接传递解码后的字符串会引发解析失败。

解决方案 #

最可靠的处理方式是将Datastore的BYTES类型转换为Base64编码字符串，同时确保BigQuery对应字段类型为BYTES，BigQuery会自动将Base64字符串解析为二进制数据。

1. 修改实体转换函数ent_to_json_func

在转换Datastore实体到JSON的逻辑中，对BYTES类型字段进行Base64编码：

import base64

def ent_to_json_func(entity, table_schema):
    json_data = {}
    # 遍历实体属性
    for prop_name, prop_value in entity.properties.items():
        if prop_value.value_type == 'BYTES':
            # 将字节数据编码为Base64字符串
            encoded_str = base64.b64encode(prop_value.bytes_value).decode('utf-8')
            json_data[prop_name] = encoded_str
        else:
            # 处理其他数据类型（如STRING、INTEGER等）的原有逻辑
            if prop_value.value_type == 'STRING':
                json_data[prop_name] = prop_value.string_value
            elif prop_value.value_type == 'INTEGER':
                json_data[prop_name] = prop_value.integer_value
            # 补充其他类型的处理逻辑
    return json_data

2. 匹配BigQuery表字段类型

确保BigQuery目标表中对应的字段类型设置为BYTES，示例Schema定义：

[
  {"name": "your_bytes_field", "type": "BYTES", "mode": "NULLABLE"},
  // 其他字段定义
]

可选方案：过滤不可打印字符（不推荐）

如果业务场景必须将字节数据转为普通字符串，可过滤掉不可打印字符，但会存在数据丢失风险：

import re

def ent_to_json_func(entity, table_schema):
    json_data = {}
    for prop_name, prop_value in entity.properties.items():
        if prop_value.value_type == 'BYTES':
            # 仅保留可打印ASCII字符
            cleaned_str = re.sub(r'[^\x20-\x7E]', '', prop_value.bytes_value.decode('utf-8', errors='ignore'))
            json_data[prop_name] = cleaned_str
        else:
            # 其他类型处理逻辑
            ...
    return json_data

验证修改 #

调整后重新运行Beam流水线，BigQuery将能正确解析Base64编码的字符串为二进制数据，避免JSON加载错误。

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