Flink SQL 解析复杂(嵌套)JSON

在 Flink 1.10 的 Table API 和 SQL 中,表支持的格式有四种:

CSV Format
JSON Format
Apache Avro Format
Old CSV Format

官网地址如下:https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#table-formats

我用 JSON Format  比较多,也有嵌套的JSON 数据需要解析,大概描述一下。

 以下内容来下官网介绍:

JSON格式允许读取和写入与给定格式 schema 相对应的JSON数据。 格式 schema 可以定义为Flink类型,JSON schema 或从所需的表 schema 派生。 Flink类型启用了更类似于SQL的定义并映射到相应的SQL数据类型。 JSON模式允许更复杂和嵌套的结构。
如果格式 schema 等于表 schema,则也可以自动派生该 schema。 这只允许定义一次 schema 信息。 格式的名称,类型和字段的顺序由表的 schema 确定。 如果时间属性的来源不是字段,则将忽略它们。 表 schema 中的from定义被解释为以该格式重命名的字段。

大概意思就是,flink 在解析json的时候,可以自己通过 schema(支持复杂的嵌套json),如果不提供 schema,默认使用 table schema 自动派生 json 的 schema(不支持复杂json)。

官网对应 json format 的表的样例:

CREATE TABLE MyUserTable (
  ...
) WITH (
  ‘format.type‘ = ‘json‘,                   -- required: specify the format type
  ‘format.fail-on-missing-field‘ = ‘true‘   -- optional: flag whether to fail if a field is missing or not, false by default

  ‘format.fields.0.name‘ = ‘lon‘,           -- optional: define the schema explicitly using type information.
  ‘format.fields.0.data-type‘ = ‘FLOAT‘,    -- This overrides default behavior that uses table‘s schema as format schema.
  ‘format.fields.1.name‘ = ‘rideTime‘,
  ‘format.fields.1.data-type‘ = ‘TIMESTAMP(3)‘,

  ‘format.json-schema‘ =                    -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
    ‘{                                      -- This also overrides the default behavior.
      "type": "object",
      "properties": {
        "lon": {
          "type": "number"
        },
        "rideTime": {
          "type": "string",
          "format": "date-time"
        }
      }
    })

注:flink 1.10 字段的名称和类型可以从 table schema 中推断,不用写  format.fields.0.name 和 format.fields.0.data-type 了。

CREATE TABLE user_log(
    user_id VARCHAR,
    item_id VARCHAR,
    category_id VARCHAR,
    behavior VARCHAR,
    ts TIMESTAMP(3)
) WITH (
    ‘connector.type‘ = ‘kafka‘,
    ‘connector.version‘ = ‘universal‘,
    ‘connector.topic‘ = ‘user_behavior‘,
    ‘connector.properties.zookeeper.connect‘ = ‘venn:2181‘,
    ‘connector.properties.bootstrap.servers‘ = ‘venn:9092‘,
    ‘connector.startup-mode‘ = ‘earliest-offset‘,
    ‘format.type‘ = ‘json‘
);

对应 json 数据如下:

{"user_id": "315321", "item_id":"942195", "category_id": "4339722", "behavior": "pv", "ts": "2017-11-26T01:00:00Z"}

 对应的字段,会映射到对应的类型上,可以直接使用,比1.9 方便了不少。

当然,这个并不是这里的主要内容。

先来个嵌套的json看下:

{"user_info":{"user_id":"0111","name":"xxx"},"timestam":1586670908699,"id":"10001"}

这样的复杂sql该怎么解析呢?

回来看下官网那段实例:

‘format.json-schema‘ =                    -- or by using a JSON schema which parses to DECIMAL and TIMESTAMP.
    ‘{                                      -- This also overrides the default behavior.
      "type": "object",
      "properties": {
        "lon": {
          "type": "number"
        },
        "rideTime": {
          "type": "string",
          "format": "date-time"
        }
      }
    }

SQL 的properties 中可以通过 属性 "format.json-schema"  设置输入的 json schema。

Flink 的 json-schema 中支持如下的数据类型:

Flink SQL 解析复杂(嵌套)JSON

 

 

 再来看下刚刚的嵌套json:

{"user_info":{"user_id":"0111","name":"xxx"},"timestam":1586670908699,"id":"10001"}

第一层的 timestam、id 直接就映射到字段上,而 user_info 也是个json。

从上面的实例上,可以看到 object 类型数据有  properties,而properties 的内容,怎么看都想是json的内层数据。

所以上面的sql 对应的 json-schema 是这样的:

‘format.json-schema‘ = ‘{
        "type": "object",
        "properties": {
           "id": {type: "string"},
           "timestam": {type: "string"},
           "user_info":{type: "object",
                   "properties" : {
                       "user_id" : {type:"string"},
                       "name":{type:"string"}
                   }
             }
        }
    }

从上面的 json schame 和 Flink SQL 的映射关系可以看出,user_info 对应的table 字段的类型是ROW,所以 table 的schema 是这样的:

CREATE TABLE user_log(
    id VARCHAR,
    timestam VARCHAR,
    user_info ROW(user_id string, name string )
)

ROW 类型的 user_info,有两个字段:user_id 和 name

注:使用的时候,直接用 "." 就可以了:如 user_info.user_id

到此,嵌套json的 schame 就搞定了。

下面我们再来看下  嵌套 json 数组:

{"user_info":{"user_id":"0111","name":"xxx"},"timestam":1586670908699,"id":"10001","jsonArray":[{"name222":"xxx","user_id222":"0111"}]}

这个又该怎么写 json schema 呢?

官网有个实例说 json format 直接解析这样的复杂 json:

 "optional_address": {
      "oneOf": [
        {
          "type": "null"
        },
        {
          "$ref": "#/definitions/address"
        }
      ]
    }

太长了,截取一段,官网明确说了支持这样的实例,也就是支持 json 数组

 json schema 和 Flink SQL 的映射关系中, json 的 array 对应 Flink SQL的 ARRAY[_]

按照 object 类型的写法,写了个这样的:

"jsonArray":{"type": "array",
             "properties": {
                      "type": "object",
                      "properties" : {
                          "user_id222" : {type:"string"},
                          "name222" : {type:"string"}
                         }
                      }
             }    

收获了一个 exception:

Caused by: java.lang.IllegalArgumentException: Arrays must specify an ‘items‘ property in node: <root>/jsonArray
    at org.apache.flink.formats.json.JsonRowSchemaConverter.convertArray(JsonRowSchemaConverter.java:264)
    at org.apache.flink.formats.json.JsonRowSchemaConverter.convertType(JsonRowSchemaConverter.java:176)
    at org.apache.flink.formats.json.JsonRowSchemaConverter.convertObject(JsonRowSchemaConverter.java:246)

然后,当然是 debug 代码了: org.apache.flink.formats.json.JsonRowSchemaConverter 就是解析 json-schema 的代码了

JsonRowSchemaConverter 类有3个主要的方法分别对应解析不同类型的数据:

// 解析 type 
private static TypeInformation<?> convertType(String location, JsonNode node, JsonNode root)
// 解析 object
private static TypeInformation<Row> convertObject(String location, JsonNode node, JsonNode root)
// 解析 array
private static TypeInformation<?> convertArray(String location, JsonNode node, JsonNode root)

convertType 方法在这里解析具体字段和类型:

for (String type : types) {
                // set field type
                switch (type) {
                    case TYPE_NULL:
                        typeSet.add(Types.VOID);
                        break;
                    case TYPE_BOOLEAN:
                        typeSet.add(Types.BOOLEAN);
                        break;
                    case TYPE_STRING:
                        if (node.has(FORMAT)) {
                            typeSet.add(convertStringFormat(location, node.get(FORMAT)));
                        } else if (node.has(CONTENT_ENCODING)) {
                            typeSet.add(convertStringEncoding(location, node.get(CONTENT_ENCODING)));
                        } else {
                            typeSet.add(Types.STRING);
                        }
                        break;
                    case TYPE_NUMBER:
                        typeSet.add(Types.BIG_DEC);
                        break;
                    case TYPE_INTEGER:
                        // use BigDecimal for easier interoperability
                        // without affecting the correctness of the result
                        typeSet.add(Types.BIG_DEC);
                        break;
                    case TYPE_OBJECT:
                        typeSet.add(convertObject(location, node, root));
                        break;
                    case TYPE_ARRAY:
                        typeSet.add(convertArray(location, node, root));
                        break;
                    default:
                        throw new IllegalArgumentException(
                            "Unsupported type ‘" + node.get(TYPE).asText() + "‘ in node: " + location);
                }
            }

简单类型,就直接添加对应的 Flink SQL 类型, 复杂类型的 object、array 由单独的方法解析,这里我们看下 covertArray:

private static TypeInformation<?> convertArray(String location, JsonNode node, JsonNode root) {
    // validate items
    if (!node.has(ITEMS)) {
        throw new IllegalArgumentException(
            "Arrays must specify an ‘" + ITEMS + "‘ property in node: " + location);
    }
    final JsonNode items = node.get(ITEMS);

    // list (translated to object array)
    if (items.isObject()) {
        final TypeInformation<?> elementType = convertType(
            location + ‘/‘ + ITEMS,
            items,
            root);
        // result type might either be ObjectArrayTypeInfo or BasicArrayTypeInfo for Strings
        return Types.OBJECT_ARRAY(elementType);
    }
    // tuple (translated to row)
    else if (items.isArray()) {
        final TypeInformation<?>[] types = convertTypes(location + ‘/‘ + ITEMS, items, root);

        // validate that array does not contain additional items
        if (node.has(ADDITIONAL_ITEMS) && node.get(ADDITIONAL_ITEMS).isBoolean() &&
                node.get(ADDITIONAL_ITEMS).asBoolean()) {
            throw new IllegalArgumentException(
                "An array tuple must not allow additional items in node: " + location);
        }

        return Types.ROW(types);
    }
    throw new IllegalArgumentException(
        "Invalid type for ‘" + ITEMS + "‘ property in node: " + location);
}

注:更多信息请查看源码(org.apache.flink.formats.json.JsonRowSchemaConverter)

从上面的代码可以看出,从 convertTypes 中解析到是 array 类型的,就调用 convertArray 方法,而 convertArray 方法中第一步就是判断是否有个 ITEMS 字段,没有直接就报错:

Arrays must specify an ‘items‘ property in node: <root>/jsonArray

有就  final JsonNode items = node.get(ITEMS)   get 出来继续解析,判断 items 是个 object 或 array (然后继续递归),都不是就抛出异常

从源码可以看出 json 数组类型的 json schema 就是这样的:

CREATE TABLE user_log(
    id VARCHAR,
    timestam VARCHAR,
    user_info ROW(user_id string, name string ),
    jsonArray ARRAY<ROW(user_id222 STRING, name222 STRING)>
) WITH (
    ‘connector.type‘ = ‘kafka‘,
    ‘connector.version‘ = ‘universal‘,
    ‘connector.topic‘ = ‘complex_string‘,
    ‘connector.properties.zookeeper.connect‘ = ‘venn:2181‘,
    ‘connector.properties.bootstrap.servers‘ = ‘venn:9092‘,
    ‘connector.startup-mode‘ = ‘earliest-offset‘,
    ‘format.type‘ = ‘json‘,
    ‘format.json-schema‘ = ‘{
        "type": "object",
        "properties": {
           "id": {type: "string"},
           "timestam": {type: "string"},
           "user_info":{type: "object",
                   "properties" : {
                       "user_id" : {type:"string"},
                       "name":{type:"string"}
                   }
             },
            "jsonArray":{"type": "array",
                         "items": {
                                  "type": "object",
                                  "properties" : {
                                      "user_id222" : {type:"string"},
                                      "name222" : {type:"string"}
                                     }
                                  }
                         }
        }
    });

看过源码之后,对于上面的json schema 就没有难度了

这里还要说下 json array 中有多个元素的案例:

{"user_info":{"user_id":"0111","name":"xxx"},"timestam":1586676835655,"id":"10001","jsonArray":[{"name222":"xxx","user_id222":"0022"},{"name333":"name3333","user_id222":"user3333"},{"cc":"xxx333","user_id444":"user4444","name444":"name4444"}]}

对应的 schema 也是这样的:

"jsonArray":{"type": "array",
                         "items": {
                                  "type": "object",
                                  "properties" : {
                                      "user_id222" : {type:"string"},
                                      "name222" : {type:"string"}
                                     }
                                  }
                         }
        }

因为在解析 json array 的时候,只能获取到一个 items 字段(多加也没用),会拿这个schema 去解析 json array 里面的所有元素,有对应字段就赋值,没用就为空

表的列也是这样的:

jsonArray ARRAY<ROW(user_id222 STRING, name222 STRING)>

在查询中直接使用 jsonArray 会将所有数据直接查出来:

INSERT INTO user_log_sink SELECT * FROM user_log;

输出的数据如下:

{"id":"10001","timestam":"1586676835655","user_info":{"user_id":"0111","name":"xxx"},"jsonArray":[{"user_id222":"0022","name222":"xxx"},{"user_id222":"user3333","name222":null},{"user_id222":null,"name222":null}]}

json array 中的第一个元素 全部解出来了,第二个元素只有 user_id222 有值,第三个元素都没解析出来

 注:json array 是这样使用的:jsonArray[1].user_id222   # 代表 jsonArray 中的第一个元素的 user_id222 字段,数组下标从 1 开始,0 或 大于实际 json array 中的 长度会报 :  java.lang.ArrayIndexOutOfBoundsException: 1 

 

欢迎关注Flink菜鸟公众号,会不定期更新Flink(开发技术)相关的推文

Flink SQL 解析复杂(嵌套)JSON

Flink SQL 解析复杂(嵌套)JSON

上一篇:通过 plsql 连接远程 Oracle


下一篇:nodejs学习笔记_nodejs和PHP在基础架构上的差别--共享状态的并发