使用JsonStreamWriter写入BigQuery JSON字段报错排查

报错根源 #

你遇到的AppendSerializationError核心原因是：BigQuery的JSON列要求输入为字符串形式的JSON数据，但当前传入的Context是嵌套JSON对象，JsonStreamWriter会默认将其解析为结构化字段，而非BQ JSON列所需的字符串格式，最终导致序列化失败。

可行解决方案 #

方案1：预处理JSON负载，将嵌套JSON转为字符串 #

把Context字段的嵌套对象序列化为JSON字符串（注意转义引号），示例修改后的负载：

[
    {
        "Context": "{\"UserAgent\": \"Mozilla/5.0 (X11; U; Linux x86_64; el-GR; rv:1.9.2.10) Gecko/20100922 Ubuntu/10.10 (maverick) Firefox/3.6.10\", \"Ip\": \"<REDACTED>\"}",
        "Channel": "browser",
        "STREAMER_DEPLOYMENT_ID": "<REDACTED>",
        // 其他字段保持不变
    }
]

这样JsonStreamWriter会把该字符串直接写入BQ的JSON列，BigQuery会自动将其解析为JSON类型。

方案2：显式指定Schema映射 #

如果不想修改原始JSON，可以通过定义显式Schema，告知JsonStreamWriter将Context字段当作字符串处理，对应到BQ的JSON列。比如在创建JsonStreamWriter时，指定包含Context为STRING类型的Schema，BigQuery会自动将字符串转换为JSON列。

额外注意点 #

• 字段名大小写：BQ表中列名为CONTEXT（全大写），而JSON负载中是Context（首字母大写），虽然BigQuery默认大小写不敏感，但建议统一字段名，避免潜在映射问题。
• 其他JSON列（如TRAITS、INTEGRATIONS）若存在类似嵌套对象，需用同样方式处理。

附原始问题信息 #

报错信息 #

com.google.cloud.bigquery.storage.v1.Exceptions$AppendSerializationError: INVALID_ARGUMENT: Append serialization failed : 
**{0=Field root.Context failed to convert to JSON. Error: JSONObject does not have a string field at root.Context.}**

原始JSON负载 #

[
    {
        "Context": {
            "UserAgent": "Mozilla/5.0 (X11; U; Linux x86_64; el-GR; rv:1.9.2.10) Gecko/20100922 Ubuntu/10.10 (maverick) Firefox/3.6.10",
            "Ip": "<REDACTED>"
        },
        "Channel": "browser",
        "STREAMER_DEPLOYMENT_ID": "<REDACTED>",
        "STREAMER_THREAD_NUMBER": 1,
        "Timestamp": "2024-10-01T11:57:37.567002Z",
        "SentAt": "2024-10-01T11:57:37.566997Z",
        "STREAMER_RUN_ID": "<REDACTED>",
        "Type": "identify",
        "Version": "1.1",
        "STREAMER_CLIENT_TIME_MS": 1727783857698,
        "STREAMER_BATCH_NUMBER": 0,
        "AnonymousId": "<REDACTED>",
        "STREAMER_INSTANCE_NUMBER": 0,
        "STREAMER_OFFSET": 0,
        "ReceivedAt": "2024-10-01T11:57:37.566945Z",
        "MessageId": "<REDACTED>"
    }
]

调试用Protobuf定义 #

message SomeMessage {
    message _context {
        string _user_agent = 1;
        string _ip = 2;
    }
    message Nested {
        _context _context = 1;
        string _channel = 2;
        string _s_t_r_e_a_m_e_r__d_e_p_l_o_y_m_e_n_t__i_d = 3;
        uint32 _s_t_r_e_a_m_e_r__t_h_r_e_a_d__n_u_m_b_e_r = 4;
        google.protobuf.Timestamp _timestamp = 5;
        google.protobuf.Timestamp _sent_at = 6;
        string _s_t_r_e_a_m_e_r__r_u_n__i_d = 7;
        string _type = 8;
        string _version = 9;
        uint64 _s_t_r_e_a_m_e_r__c_l_i_e_n_t__t_i_m_e__m_s = 10;
        uint32 _s_t_r_e_a_m_e_r__b_a_t_c_h__n_u_m_b_e_r = 11;
        string _anonymous_id = 12;
        uint32 _s_t_r_e_a_m_e_r__i_n_s_t_a_n_c_e__n_u_m_b_e_r = 13;
        uint32 _s_t_r_e_a_m_e_r__o_f_f_s_e_t = 14;
        google.protobuf.Timestamp _received_at = 15;
        string _message_id = 16;
    }
    repeated Nested items = 1;
}

BigQuery表结构 #

CREATE TABLE TEST.EVENTS (
    STREAMER_THREAD_NUMBER INT,
    MESSAGEID STRING(255),
    STREAMER_BATCH_NUMBER INT,
    STREAMER_INSTANCE_NUMBER INT,
    STREAMER_OFFSET BIGINT,
    STREAMER_CLIENT_TIME_MS INT,
    CONTEXT JSON,
    ANONYMOUSID STRING(255),
    CHANNEL STRING(255),
    RECEIVEDAT STRING(255),
    TIMESTAMP STRING(255),
    SENTAT STRING(255),
    USERID STRING(255),
    VERSION STRING(255),
    STREAMER_RUN_ID STRING(255),
    STREAMER_DEPLOYMENT_ID STRING(255),
    TRAITS JSON,
    TYPE STRING(255),
    INTEGRATIONS JSON,
    PROPERTIES JSON,
    EVENT STRING(255),
    ORIGINALTIMESTAMP STRING(255)
);

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