Flink TableFunction 解析嵌套 JSON 数组：从“UNNEST 报错”到“STRING 内层解析”

消息体 (JSON object)
├─ data      : ARRAY<ROW< … > >      ← 外层是真数组，需要 UNNEST 一次
│   └─ 每个 ROW 有二十余列，其中一列是：
│       account_detail : STRING      ← 关键：是“字符串”，里面塞着转义的 JSON 数组
├─ old       : ARRAY<ROW< … > >      （旧值，本场景可忽略）
├─ type      : STRING   ("UPDATE")
├─ database  : STRING
└─ ts        : BIGINT

[ { "field1": "...", "field2": "..." },
  { "field1": "..." },
  ... ]                       ← 这是“被当作字符串存的 JSON 数组”

-- 把 account_detail 声明成 ARRAY<ROW<...>>，然后两层 UNNEST
SELECT u.uid, b.field1, b.field2
FROM user_source AS S
CROSS JOIN UNNEST(S.data) AS u                              -- 摊开外层
CROSS JOIN UNNEST(u.account_detail) AS b                     -- 想摊开内层

org.apache.flink.formats.json.JsonParseException: Fail to deserialize at field: data.
  at org.apache.flink.formats.json.JsonParserToRowDataConverters
       .lambda$createRowConverter$43b82837$1(...:419)
  at org.apache.flink.formats.json.JsonParserToRowDataConverters
       .lambda$wrapIntoNullableConverter$ca96cb8f$1(...:463)
  at org.apache.flink.formats.json.JsonParserRowDataDeserializationSchema
       .deserialize(...:91)
  ...
Caused by: java.io.IOException: Failed to deserialize JSON
'{
  "data": [ { "uid": <int>, ... , "account_detail": "[]", ... } ],
  "old"   : [ { ... } ],
  "type"  : "UPDATE",
  "database": "<db>",
  "table" : "<table>",
  "ts"    : <long>,
  "pkNames": [ "<pk>" ]
}'
  at org.apache.flink.formats.json.JsonParserRowDataDeserializationSchema
       .deserialize(...:97)
  at org.apache.flink.streaming.connectors.kafka.table
       .DynamicKafkaDeserializationSchema.deserialize(...:113)
  ...
Caused by: org.apache.flink.formats.json.JsonParseException:
  Fail to deserialize at field: data.

UNNEST requires a MULTISET / ARRAY type, but got STRING

CREATE TABLE user_sensitive_source (
    data ARRAY<ROW<
        uid            BIGINT,
        /* ... 其余通用属性省略 ... */
        account_detail STRING          -- 关键：内层 JSON 数组以“字符串”形式存放
        /* ... 其余属性省略 ... */
    >>,
    type      STRING,
    ts        BIGINT,
    proctime AS PROCTIME()
) WITH (
     'connector' = 'kafka',
     'topic'     = 'user_sensitive_topic',
     -- ${kafka_bootstrap} / 凭据等环境相关配置用变量注入，不写死
     'scan.startup.mode' = 'latest-offset',
     'properties.auto.offset.reset' = 'earliest',
     'format' = 'json'
);

-- 第一层：UNNEST 把外层 ARRAY<ROW> 摊成“一行一个用户”
CREATE TEMPORARY VIEW v_user AS
SELECT u.uid, u.account_detail
FROM user_sensitive_source AS S
CROSS JOIN UNNEST(S.data) AS u
WHERE u.uid IS NOT NULL;

-- 第二层：内层是 STRING，交给表函数展开；LEFT JOIN 保留“无子记录”用户
CREATE TEMPORARY VIEW v_user_unnest AS
SELECT
    u.uid,
    b.field1, b.field2, b.field3
FROM v_user AS u
LEFT JOIN LATERAL TABLE(parse_detail(u.account_detail)) b ON TRUE
WHERE u.uid IS NOT NULL;

package com.fb.udf;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

// 三个 Topic / 三套 Schema 共用这一个类：
// 永远“按键取值、缺失即 null”；别名做 fallback 统一到同一列；
// 输出固定 N 列，方便多 Topic 之后直接 UNION ALL。
@FunctionHint(
        input = @DataTypeHint("STRING"),
        output = @DataTypeHint(
                   "ROW<field1 STRING, field2 STRING, field3 STRING>"
         )
)
public class AccountDetailParserYH extends TableFunction<Row> {

    private static final long serialVersionUID = 1L;
    private static final ObjectMapper MAPPER = new ObjectMapper();

    public void eval(String detail) {
        if (detail == null) {
            return;
         }

        String value = unquote(detail.trim());

         // 空串 / 空数组 / 字面量 null：一行都不产出
        if (value.isEmpty()
                 || "[]".equals(value)
                 || "null".equalsIgnoreCase(value)) {
            return;
         }

        try {
            JsonNode root = MAPPER.readTree(value);

             // 外层可能被二次转义成字符串，再解析一层
            if (root.isTextual()) {
                String inner = root.asText();
                if (inner == null || inner.trim().isEmpty()) {
                    return;
                 }
                root = MAPPER.readTree(inner);
             }

            if (!root.isArray()) {
                throw new IllegalArgumentException(
                         "detail must be a JSON array, actual: " + value
                 );
             }

            for (JsonNode item : root) {
                if (item == null || item.isNull()) {
                    continue;
                 }
                if (!item.isObject()) {
                    throw new IllegalArgumentException(
                             "array element must be JSON object, actual: " + item
                     );
                 }

                 // 字段 A 缺失时回退到 B，落到同一列
                String fallback = firstNonNull(
                        getAsString(item, "x"),
                        getAsString(item, "x2"));

                Row row = new Row(3);
                row.setField(0, getAsString(item, "field1"));
                row.setField(1, getAsString(item, "field2"));
                row.setField(2, getAsString(item, "field3"));

                collect(row);
             }
         } catch (Exception e) {
            throw new RuntimeException("Failed to parse detail JSON: " + value, e);
         }
     }

    private static String firstNonNull(String a, String b) {
        return a != null ? a : b;
     }

     // 剥掉首尾的单/双引号
    private static String unquote(String s) {
        if (s.length() >= 2
                 && ((s.charAt(0) == '\'' && s.charAt(s.length() - 1) == '\'')
                  || (s.charAt(0) == '"' && s.charAt(s.length() - 1) == '"'))) {
            s = s.substring(1, s.length() - 1);
         }
        return s;
     }

    private String getAsString(JsonNode node, String fieldName) {
        JsonNode field = node.get(fieldName);
        if (field == null || field.isNull()) {
            return null;
         }
         // 容器（对象/数组）取 JSON 串，标量取文本
        if (field.isContainerNode()) {
            return field.toString();
         }
        return field.asText();
     }
}

ADD CUSTOMJAR 'rs:udf/account-detail-parser-1.0.jar';

CREATE TEMPORARY SYSTEM FUNCTION parse_detail
  AS 'com.fb.udf.AccountDetailParserYH';

CREATE TABLE dw_user_detail_sink (
    uid      VARCHAR(32),
    field1   VARCHAR(128),
    field2   VARCHAR(128),
    field3   VARCHAR(128),
    PRIMARY KEY (uid) NOT ENFORCED
) WITH (
     -- 'connector' = 'print'   -- 调试期先打 stdout
     -- 生产切 jdbc：'connector'/'url'/'table-name'/'driver' 等用 ${...} 注入
     'connector' = 'print'
);

INSERT INTO dw_user_detail_sink
SELECT
    CAST(b.uid AS VARCHAR)     AS uid,
    COALESCE(b.field1, '')     AS field1,
    COALESCE(b.field2, '')     AS field2,
    COALESCE(b.field3, '')     AS field3
FROM v_user_unnest b;