文章目录一. 需求描述二. 方案思路1. 解决思路2. flink json 解析2.1. 通过json path解析非array数据2.2. 通过json path解析array数据3. CROSS JOIN逻辑三. 方案实现1. http json数据样例2. flink sql 说明一. 需求描述flink消费http接口的数据将json中的数组展开多行如下样例数据以及要求处理的数据效果{name:John Doe,age:30,address:{street:{street:123 Main St,city:New York,state:NY},city:New York,state:NY},phoneNumbers:[{type:home,number:212-555-1234},{type:fax,number:646-555-4567}],children:[],spouse:null}nameagestreetcitystatephone_typephone_numberJohn Doe30123 Main StNew YorkNYhome212-555-1234John Doe30123 Main StNew YorkNYfax646-555-4567二. 方案思路1. 解决思路flink 消费http接口的数据json发送到下游下游算子解析json数据当遇到数组时算子解析返回array通过使用CROSS JOIN 将数组数据拍平如上表格展现2. flink json 解析2.1. 通过json path解析非array数据如下通过flink内置函数JSON_VALUE 进行数据解析支持多种类型的输出默认输出为string。这里使用 cast转换如下举例cast(JSON_VALUE(json_string,$.id)asint)asid,JSON_VALUE(json_string,$.name)asname,cast(JSON_VALUE(json_string,$.details.age.real)asint)asreal,JSON_VALUE(json_string,$.details.address)asaddress,2.2. 通过json path解析array数据官网目前JSON_QUERY虽然能够包装为array但实际上总是会返回为string不符合要求。如下dependenciesdependencygroupIdcom.jayway.jsonpath/groupIdartifactIdjson-path/artifactIdversion2.6.0/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-common/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency通过udf解决packagecom.dtstack.chunjun.local.test;importcom.jayway.jsonpath.JsonPath;importorg.apache.flink.table.functions.ScalarFunction;importjava.util.ArrayList;importjava.util.List;publicclassJsonArrayFieldExtractorextendsScalarFunction{publicListStringeval(StringjsonString,StringjsonPath){if(jsonStringnull||jsonString.isEmpty()){returnnewArrayListString();}try{List?resultJsonPath.read(jsonString,jsonPath);ListStringstringListnewArrayList();for(Objectobj:result){stringList.add(obj.toString());}returnstringList;}catch(Exceptione){returnnewArrayListString();}}}3. CROSS JOIN逻辑Array Expansion注意CROSS JOIN 返回两个连接表的笛卡尔积当有多个数组时会产生笛卡尔积。比如两个数组分别有100个元素那么如果使用两次CROSS JOIN 则会产生1万行数据。三. 方案实现1. http json数据样例{id:1,name:Alice,details:{age:{real:11},address:123Mainst,contacts:[{type:email,value:aliceexample.com},{type:phone,value:123-456-7890}],grade:[{grade:[{zz:11},{zz:11}],bb:{rr:{yy:aliceexample.com}}},{grade:[{zz:22}],bb:{rr:{yy:aliceexample.com}}}]}}2. flink sql 说明CREATETEMPORARYSYSTEMFUNCTIONget_json_arrayAScom.dtstack.chunjun.local.test.JsonArrayFieldExtractor;CREATETABLEsource(json_stringvarchar)WITH(connectorhttp-x,urlhttp://localhost:8088/api/arraypage,intervalTime3000,methodget--请求方式get 、post,decodetext-- 数据格式只支持json模式-- 以下4个参数要同时存在,page-param-namepagenum-- 多次请求参数1分页参数名例如pageNum,start-index1-- 多次请求参数2开始的位置,end-index4-- 多次请求参数3结束的位置,step1-- 多次请求参数4步长默认值为1);CREATETABLEsink(idint,namevarchar,realint,addressvarchar,zzint,yyvarchar)WITH(connectorprint);insertintosinkSELECTcast(JSON_VALUE(json_string,$.id)asint)asid,JSON_VALUE(json_string,$.name)asname,cast(JSON_VALUE(json_string,$.details.age.real)asint)asreal,JSON_VALUE(json_string,$.details.address)asaddress,cast($.grade[*].grade[*].zzasint)aszz,$.details.grade[*].bb.rr.yyasyyFROMsourceCROSSJOINUNNEST(get_json_array(json_string,$.details.grade[*].grade[*].zz))AST($.grade[*].grade[*].zz)CROSSJOINUNNEST(get_json_array(json_string,$.details.grade[*].bb.rr.yy))AST1($.details.grade[*].bb.rr.yy);--{-- id: 1,-- name: Alice,-- details: {-- age: {real:11},-- address: 123Mainst,-- contacts: [-- {-- type: email,-- value: aliceexample.com-- },-- {-- type: phone,-- value: 123-456-7890-- }-- ],-- grade: [-- {-- grade: [{zz:11},{zz:11}],-- bb: {rr:{yy:aliceexample.com}}-- },-- {-- grade: [{zz:22}],-- bb: {rr:{yy:aliceexample.com}}-- }-- ]-- }--}消费结果具体逻辑描述http连接器消费http接口数据 具体使用chunjun的http连接器相关代码见我提供的相关pr[feature-DTStack#1775][connector][http] http supports offline mode使用JSON_VALUE、get_json_array解析为string和arraystring,之后使用cast进行类型转换CROSS JOIN 生成笛卡尔积