收到需求后我第一反应是这不就是实时数仓里最常见的活吗Kafka里推过来一条消息外层套个公共字段里面又套了业务对象业务对象下面还有数组数组元素里又是一个多层的结构体。要是字段都是平铺的Flink SQL直接映射字段就完了难就难在它是“复杂JSON”。标题里提到的场景本质上是两件事第一用Flink消费Kafka里的复杂JSON第二Flink SQL没有类似Spark那种好用的get_json_object函数得自己写一个UDF把嵌套字段提取出来方便继续加工和打印调试。先说结论这个方案我完整跑通过核心就是自己实现一个基于JSONPath思路的ScalarFunction注册成get_json_object后在Flink SQL里直接用。整个链路是Kafka → Flink Table API/SQL → 自定义UDF解析 → print连接器输出。下面把设计思路、函数实现、建表配置、常见坑一次性说清楚适合正在用Flink做日志解析、埋点清洗、实时数仓的同学直接参考。1. 项目整体设计与思路拆解为什么选Flink Kafka这套组合1.1 需求核心复杂JSON到底复杂在哪先看一条实际开发中很典型的消息。上游系统上报用户行为消息体长这样{ common: { appId: mall_app, userId: u_10001, ts: 1713177600000, device: { os: android, brand: xiaomi } }, event: { eventId: order_submit, properties: { orderAmount: 99.8, items: [ { skuId: s_8899, num: 1, price: 99.8, attrs: { color: black, size: L } } ] } }, traceId: trace_xxx_123 }这种结构不是单层KV而是典型的对象嵌套对象、对象嵌套数组、数组元素再裹对象。如果用Kafka JSON Format直接映射到Flink Tableschema得提前写清楚每一层的字段路径而且数组元素里的结构还得定义成ROW类型或ARRAY 。上游字段一旦增删作业就得改某些字段值类型不稳定比如orderAmount有时候是数字有时候是字符串JSON Format解析还会直接报错。这在真实业务里非常常见也是我优先考虑用自定义函数解析的根本原因。1.2 为什么不直接用Flink SQL内置JSON函数Flink Table SQL里其实有个JSON_VALUE函数能从JSON字符串中提取值。但实测下来有几个非常别扭的地方第一JSON_VALUE返回类型是STRING走的是SQL标准JSON路径语法对$.event.properties.items[0].skuId这种带数组下标的路径支持不完整不同版本行为还有差异。第二它对“路径不存在”和“JSON格式不合法”的处理比较死板默认会抛异常需要自己加ON EMPTY、ON ERROR这类子句才能兜底。第三缺失字段返回空字符串而不是NULL后续做聚合和判断时容易埋雷。第四它不擅长提取整个子对象。比如我想把device这个对象整体取出来打成一行调试JSON_VALUE做不到JSON_QUERY可以做但语义和返回类型一样有坑。对比之下Spark SQL里get_json_object(json_str, $.path)是很多数仓工程师用惯了的路径写法直观、路径不存在返回NULL、能返回子对象序列化后的字符串。可惜Flink没有内置等价函数。所以“自定义一个get_json_object”不是没事找事而是把Spark时代的SQL习惯平滑迁移到Flink同时补齐内置函数的短板。1.3 整体技术路线raw格式接住整条JSONUDF负责一切解析这个方案里有一个关键选型我强烈建议初次做的人注意Kafka源表不要用format json用format raw。用JSON Format时Flink解析器会把Kafka消息按你定义的schema映射成RowData。如果消息里多了一个字段默认会报错或者忽略如果字段类型不匹配比如定义成INT实际来的是字符串作业会直接脏数据。这个模式下复杂JSON反而成了负担schema管理极其痛苦。用RAW Format时整条消息就是一个STRING字段我把完整的JSON字符串交给自定义UDF路径提取完全在函数内部处理。好处是三句话能说清上游加字段完全不用改作业解析逻辑在函数里字段类型不合法不会让作业崩函数内部做了兜底调试阶段想打印整条原始JSON也可以直接透传非常直观。所以整体思路就是RAW收数据UDF做解析PRINT做输出。这个链路简单、可控、易排查适合作为复杂JSON接入Flink的第一版方案。2. 环境准备与Kafka接入的配置细节2.1 项目依赖和版本对齐我用的环境是Flink 1.171.16、1.18也适用主要注意connector版本匹配。Maven项目里核心依赖长这样properties flink.version1.17.2/flink.version scala.binary.version2.12/scala.binary.version /properties dependencies !-- Flink Table API -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- Kafka Connector -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version${flink.version}/version /dependency !-- JSON解析库这里用fastjson2纯粹是JSONPath好用 -- dependency groupIdcom.alibaba.fastjson2/groupId artifactIdfastjson2/artifactId version2.0.43/version /dependency /dependencies版本上有几个坑值得提前说flink-table-api-java-bridge用providedscope打包时不要打进去否则跟运行时自带的Table API冲突大概率报NoSuchMethodError。Kafka connector的版本必须和Flink主版本对应比如Flink 1.17就用flink-connector-kafka:1.17.2不要自己乱拉别的版本。如果用SQL Client提交UDF打成fat jar放进lib目录或者用-C参数指定依赖路径fastjson2也必须在classpath里。2.2 Kafka源表RAW模式才是复杂JSON的正确姿势建源表时我推荐写成这样CREATE TABLE kafka_source ( raw_json STRING ) WITH ( connector kafka, topic user_behavior_topic, properties.bootstrap.servers localhost:9092, properties.group.id flink_json_debug_group, properties.auto.offset.reset earliest, format raw );重点在这个format raw。很多人习惯性写成format json然后发现parse失败。原因很简单JSON format要求你显式声明每个字段同时它会把Kafka消息当JSON对象解析而复杂JSON里某个字段是数字还是字符串Kafka消息本身并没有schema信息解析器只能按你的表结构猜。猜对了是运气猜错了作业抛异常。RAW模式就没有这个问题Kafka里消息字节流是什么样raw_json字段就是什么样永远不解析失败。这才是处理“不规整复杂JSON”的最稳入口。2.3 打印输出表PRINT连接器的配置要点打印目标表我用内置PRINT连接器CREATE TABLE print_sink ( user_id STRING, event_id STRING, sku_id STRING, os STRING, raw_snapshot STRING ) WITH ( connector print );PRINT连接器会把每一条结果以I[字段1, 字段2, ...]的形式写到TaskManager的标准输出里查日志就能看到。这里有个非常实用的建议调试阶段务必把作业并行度设为1否则多个subtask同时往stdout打数据你根本对不上行与行的顺序看着像乱序其实是并行输出交错。SET parallelism.default 1;如果用的是DataStream API可以这样设置执行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1);打印虽然是调试期最常用的手段但它对结果集没有汇聚能力数据量大时日志会刷得飞快。想要控制打印量可以在SQL里加LIMIT仅批/TableResult模式支持或者配合HOP窗口、TUMBLE窗口只跑一小段。后面问题排查章节我再细说。3. 自定义get_json_object函数的核心实现3.1 ScalarFunction的基本骨架Flink Table API里实现用户自定义函数最简单的是继承ScalarFunction重写eval方法。它的基本规则是一个UDF类里可以有多个eval重载方法的第一个参数默认代表函数的第一个入参。我完整的实现如下使用fastjson2内置的JSONPath支持import org.apache.flink.table.functions.ScalarFunction; import org.apache.flink.table.annotation.DataTypeHint; import org.apache.flink.table.annotation.InputGroup; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONPath; public class GetJsonObject extends ScalarFunction { DataTypeHint(value STRING, bridgedTo String.class) public String eval( DataTypeHint(inputGroup InputGroup.ANY) Object jsonField, String jsonPath) { if (jsonField null || jsonPath null) { return null; } String jsonStr; if (jsonField instanceof String) { jsonStr (String) jsonField; } else { jsonStr jsonField.toString(); } if (jsonStr.trim().isEmpty()) { return null; } try { Object result JSONPath.of(jsonPath).extract(JSON.parse(jsonStr)); if (result null) { return null; } return result instanceof String ? (String) result : JSON.toJSONString(result); } catch (Exception e) { return null; } } }这段代码有几个设计是刻意为之的逐个说明。3.2 为什么用JSONPath而不是手动split解析自己写JSON解析最自然的想法是先parse成JSONObject然后一层层get。比如JSON.parseObject(jsonStr).getJSONObject(event).getJSONObject(properties).getJSONArray(items).getJSONObject(0).getString(skuId)。写起来不复杂但问题在于路径一变就得改代码。真实业务里JSON结构可能有几十个字段路径每条路径写一长串getCoder很容易麻。JSONPath的好处是把“路径”和“解析逻辑”解耦。代码里只需要传$.event.properties.items[0].skuId这种字符串函数内部统一处理。因为fastjson2的JSONPath支持$.common.userId取子节点$.event.properties.items[0]数组按下标取元素$.event.properties.items[*].skuId取数组所有元素的skuId字段返回数组$.common.device取整个子对象核心逻辑就一行JSONPath.of(jsonPath).extract(JSON.parse(jsonStr))。路径不对就抛异常函数捕获后返回NULL。这已经把八成需求都覆盖了。3.3 参数类型为什么要设计成Object Stringeval方法的第一个参数类型我没写死成String而是用DataTypeHint(inputGroup InputGroup.ANY)标注成Object再用注解指定输出是STRING。原因是我在实验中发现Flink SQL调用UDF时如果传入字段本身是STRING类型没问题但有些场景下用户会用CAST(raw_json AS STRING)后的结果再传给函数或者字段在Table层已经被声明成其他类型此时如果eval只接受String可能因为类型不匹配导致注册或调用失败。统一收成Object内部做String转换兼容性最好。实际开发我也建议所有“解析类”UDF都这样设计别为了省事把入参类型收窄。3.4 返回子对象和数组的策略get_json_object在Spark里有个行为值得抄过来如果路径指向的是一个JSON对象或数组返回的是对象序列化后的字符串而不是报错。上面的代码就是照这个行为做的return result instanceof String ? (String) result : JSON.toJSONString(result);比如传入$.common.device返回的就是{os:android,brand:xiaomi}这个JSON字符串。调试时想确认某个子结构的完整内容这一招特别管用。下游如果还想继续提里面的字段可以再调一次get_json_object形成链式解析。3.5 容错解析失败永远返回NULL而不是抛异常流式作业最忌讳一个坏数据让整个任务崩掉。Kafka里偶尔混入一条非JSON字符串、字段类型莫名其妙、路径因为上游版本变更失效都是家常便饭。所以函数里所有解析逻辑都包在try-catch里异常捕获后直接返回NULL。这个策略导致的结果就是解析不出数据时你不会看到报错只会看到字段为NULL。排查时就需要靠日志和审计这也是后面问题章节要讲的点。生产上我建议在UDF之上做一个配套的“原始数据监控”比如记录raw_json为NULL的比例一旦飙升就去查上游。3.6 注册函数并测试Flink SQL里注册临时函数的方法很简单CREATE TEMPORARY FUNCTION get_json_object AS com.example.udf.GetJsonObject;如果是通过TableEnvironment注册TableEnvironment tEnv TableEnvironment.create(EnvironmentSettings.inStreamingMode()); tEnv.createTemporarySystemFunction(get_json_object, GetJsonObject.class);这里有个容易翻车的细节CREATE TEMPORARY SYSTEM FUNCTION创建的是系统级函数所有database通用CREATE TEMPORARY FUNCTION创建的是会话级。在SQL Client里用CREATE TEMPORARY FUNCTION就够了但要注意函数类名必须带全限定名且JAR已经在classpath中。注册完先别急着接Kafka我习惯先在SQL Client里跑一个最简单的验证SELECT get_json_object( {common:{userId:u_10001},items:[{skuId:s_1}]}, $.items[0].skuId );能正确输出s_1再接Kafka联调省得两头排查。4. 完整实战从Kafka复杂JSON到控制台打印4.1 先造一条有代表性的复杂JSON为了演示完整链路我们往Kafka topicuser_behavior_topic里生产一条跟开头类似的数据这里再贴一遍后面建表和分析都基于它{ common: { appId: mall_app, userId: u_10001, ts: 1713177600000, device: {os: android, brand: xiaomi} }, event: { eventId: order_submit, properties: { orderAmount: 99.8, items: [ {skuId: s_8899, num: 1, price: 99.8, attrs: {color: black, size: L}} ] } }, traceId: trace_xxx_123 }生产消息的时候可以用Kafka自带命令行工具也可以用任意客户端脚本。关键是消息的value是上面这串JSON的原始文本不要加多余的前后缀。4.2 建表、创建函数、写查询完整跑通的SQL脚本如下-- 1) 源表raw格式拿整条JSON CREATE TABLE kafka_source ( raw_json STRING ) WITH ( connector kafka, topic user_behavior_topic, properties.bootstrap.servers localhost:9092, properties.group.id flink_json_debug_group, properties.auto.offset.reset earliest, format raw ); -- 2) 注册自定义UDF CREATE TEMPORARY FUNCTION get_json_object AS com.example.udf.GetJsonObject; -- 3) 打印目标表 CREATE TABLE print_sink ( user_id STRING, event_id STRING, first_sku_id STRING, os STRING, device_json STRING ) WITH ( connector print ); -- 4) 提取字段并打印 INSERT INTO print_sink SELECT get_json_object(raw_json, $.common.userId) AS user_id, get_json_object(raw_json, $.event.eventId) AS event_id, get_json_object(raw_json, $.event.properties.items[0].skuId) AS first_sku_id, get_json_object(raw_json, $.common.device.os) AS os, get_json_object(raw_json, $.common.device) AS device_json FROM kafka_source;查询逻辑里我特意放了两个不同类型的提取一个是原子字段一个是整个子对象。前者返回像u_10001这种普通字符串后者返回{os:android,brand:xiaomi}这种JSON字符串。如果你调试时需要把整条JSON也打出来直接在SELECT里加上raw_json字段就好。4.3 运行结果与日志解读作业跑起来后在TaskManager日志或者stdout里可以看到类似这样的输出I[u_10001, order_submit, s_8899, android, {os:android,brand:xiaomi}]I表示这条是Insert操作后面括号里的值就是SELECT语句定义的五个字段顺序一一对应。如果某个字段路径写错比如把$.event.properties.items[1].skuId当成路径下标1不存在对应位置就是null其它字段正常输出。这一点可以反向验证UDF容错逻辑是否生效。实时流里PRINT输出是没有终点的只要Kafka里持续有数据stdout就会一直刷。做一次性验证时建议用kafka-console-producer手动生产几条消息然后启动作业观察几秒确认结果后停掉作业。如果非要用自动生成的数据量大一些最好加个窗口或者LIMIT限制输出量。4.4 并行度对打印结果的影响前文提过打印调试时并行度设为1最省心。假如并行度是4Kafka分区有多个每个subtask都会独立打印自己消费的数据日志里会出现多个I[穿插在一起结果看起来就像错乱。这不是数据错乱是打印并发造成的假象。如果你只是想验证解析逻辑而不是验证并行处理能力先把并行度设为1再做打印。等到真正上线再把并行度调大并把PRINT sink换成Kafka或目标存储。5. 常见问题与排查技巧实录5.1 解析结果全是NULL这是最高频的问题原因分布我整理成表格可能原因具体表现排查方法路径写错所有目标字段都为NULL在SQL Client里用一条固定JSON试路径字段名大小写不一致只有部分字段为NULL对照原始JSON逐字符核对字段名源表不是raw格式作业直接失败或字段解析错乱检查WITH里format是不是rawJSON字符串前后有空白偶发解析失败在UDF里trim后再parse上游消息不是合法JSON整行NULL单独打印raw_json检查要避免这个问题的最高效手段一定是先在离线环境里把UDF配合固定JSON测一遍。路径写错是最常见错误比如把JSON里的字段名userId写成userid或者下标从1开始而不是0。这些都是肉眼难查的坑靠SQL Client单测几秒钟就能暴露。5.2 JSON字段名包含特殊字符怎么办有些JSON的key不是标准字母数字比如user-id、a.b这种。用$.common.user-id这种路径JSONPath会把它当成user减id解析结果自然不对。fastjson2在这种场景下支持把key用双引号括起来get_json_object(raw_json, $.user-id)我印象中这在Spark的get_json_object里也类似。平时写路径时如果发现解析结果不对先想是不是key里有特殊字符。这个细节很容易被忽略一旦踩中会浪费很多时间。5.3 打印数据量太大刷爆日志PRINT连接器默认是每条数据都输出数据量大时TaskManager的日志文件会快速膨胀最终影响磁盘和排查。调试阶段我通常这样控制在Kafka生产端只发几条测试消息或者源表加上scan.startup.mode earliest-offset并且用一次性运行跑完取消作业或者SQL里构造窗口比如只统计最近一小段窗口内的数据再打印严格意义上PRINT连接器没有“限流”参数控制输出量要靠作业本身。如果你只是临时验证最快的方式是启动作业前先用命令行工具确认topic里数据条数再决定要不要跑。5.4 UDF注册不生效常见现象是SQL Client执行CREATE TEMPORARY FUNCTION时报ClassNotFoundException或者作业启动时函数找不到。通常是JAR没打进去或者类名拼错。用SQL Client时启动命令要带上JARflink sql-client -j flink-udf.jar如果用-C参数指定目录要确认目录下真的有这个jar。注意SQL Client里的函数是会话级的重启SQL Client后需要重新注册。生产环境建议用Catalog持久化函数或者直接在作业代码里注册避免每次手动操作。5.5 fastjson2和其他JSON库的依赖冲突Flink本身带了一些Jackson相关依赖UDF里用fastjson2一般没事但如果项目中同时引入了多个JSON库可能出现版本冲突导致奇奇怪怪的NoSuchMethodError。我的做法是UDF所在的JAR使用shade插件打包把fastjson2重定位relocation到自定义包名避免跟运行时依赖冲突。下面是一个简化版shade配置plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration relocations relocation patterncom.alibaba.fastjson2/pattern shadedPatterncom.example.shaded.fastjson2/shadedPattern /relocation /relocations filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters /configuration /execution /executions /plugin5.6 性能与吞吐JSONPath解析会不会成为瓶颈有人会担心每条数据都做一次JSON parse和JSONPath extract吞吐上不去。实测下来单条消息几百字节的规模单并行度每秒处理几千条完全没问题。真正需要高性能的场景可以从两个方向优化复用JSONPath实例不要每次都of(jsonPath)提前编译好常用路径对象传入的字符串如果可以直接让上游规范成紧凑JSON减少序列化开销。但说句实在话对于绝大多数日志解析、埋点清洗场景自定义UDF的解析性能不是瓶颈Kafka的消费能力和下游存储往往才是。6. 个人实操经验工具箱越简单越可靠整个项目做完我个人最深的体会是处理Flink里的复杂JSON不要一上来就整复杂的schema推导或类型映射。先用RAW模式接住字符串再配合一个灵活的JSONPath提取函数是最快、最稳、最好排查的路径。你可能最后发现这个UDF占了80%的解析工作后面就算要接正经的schema映射也可以在这个函数返回结果的基础上去做。另外一个小技巧给这个UDF加一个“原始数据透传”重载方法。比如eval(String json, String path)和eval(String json)后者直接返回原始JSON字符串。调试的时候把raw_json和解析字段一起打印出问题一眼就能定位是解析逻辑的问题还是上游数据格式的问题。这个自定义函数的模式还可以继续扩展。比如把返回的JSON字符串再做一次CAST AS或JSON_QUERY操作就能满足更复杂的后续加工或者把这个函数放进持久化Catalog团队其他人直接SELECT get_json_object(x, $.y)就能用不用每个人都重写一遍。处理Kafka复杂JSON这件事本质上不是Flink能力不够而是你需要一个趁手的工具。get_json_object就是那个工具花半小时写完后面能省无数个小时。 
长河君 · 2026/10/14 22:31:18 · 阅读 1.2k Flink SQL自定义UDF解析Kafka复杂JSON完整实战
收到需求后我第一反应是这不就是实时数仓里最常见的活吗Kafka里推过来一条消息外层套个公共字段里面又套了业务对象业务对象下面还有数组数组元素里又是一个多层的结构体。要是字段都是平铺的Flink SQL直接映射字段就完了难就难在它是“复杂JSON”。标题里提到的场景本质上是两件事第一用Flink消费Kafka里的复杂JSON第二Flink SQL没有类似Spark那种好用的get_json_object函数得自己写一个UDF把嵌套字段提取出来方便继续加工和打印调试。先说结论这个方案我完整跑通过核心就是自己实现一个基于JSONPath思路的ScalarFunction注册成get_json_object后在Flink SQL里直接用。整个链路是Kafka → Flink Table API/SQL → 自定义UDF解析 → print连接器输出。下面把设计思路、函数实现、建表配置、常见坑一次性说清楚适合正在用Flink做日志解析、埋点清洗、实时数仓的同学直接参考。1. 项目整体设计与思路拆解为什么选Flink Kafka这套组合1.1 需求核心复杂JSON到底复杂在哪先看一条实际开发中很典型的消息。上游系统上报用户行为消息体长这样{ common: { appId: mall_app, userId: u_10001, ts: 1713177600000, device: { os: android, brand: xiaomi } }, event: { eventId: order_submit, properties: { orderAmount: 99.8, items: [ { skuId: s_8899, num: 1, price: 99.8, attrs: { color: black, size: L } } ] } }, traceId: trace_xxx_123 }这种结构不是单层KV而是典型的对象嵌套对象、对象嵌套数组、数组元素再裹对象。如果用Kafka JSON Format直接映射到Flink Tableschema得提前写清楚每一层的字段路径而且数组元素里的结构还得定义成ROW类型或ARRAY 。上游字段一旦增删作业就得改某些字段值类型不稳定比如orderAmount有时候是数字有时候是字符串JSON Format解析还会直接报错。这在真实业务里非常常见也是我优先考虑用自定义函数解析的根本原因。1.2 为什么不直接用Flink SQL内置JSON函数Flink Table SQL里其实有个JSON_VALUE函数能从JSON字符串中提取值。但实测下来有几个非常别扭的地方第一JSON_VALUE返回类型是STRING走的是SQL标准JSON路径语法对$.event.properties.items[0].skuId这种带数组下标的路径支持不完整不同版本行为还有差异。第二它对“路径不存在”和“JSON格式不合法”的处理比较死板默认会抛异常需要自己加ON EMPTY、ON ERROR这类子句才能兜底。第三缺失字段返回空字符串而不是NULL后续做聚合和判断时容易埋雷。第四它不擅长提取整个子对象。比如我想把device这个对象整体取出来打成一行调试JSON_VALUE做不到JSON_QUERY可以做但语义和返回类型一样有坑。对比之下Spark SQL里get_json_object(json_str, $.path)是很多数仓工程师用惯了的路径写法直观、路径不存在返回NULL、能返回子对象序列化后的字符串。可惜Flink没有内置等价函数。所以“自定义一个get_json_object”不是没事找事而是把Spark时代的SQL习惯平滑迁移到Flink同时补齐内置函数的短板。1.3 整体技术路线raw格式接住整条JSONUDF负责一切解析这个方案里有一个关键选型我强烈建议初次做的人注意Kafka源表不要用format json用format raw。用JSON Format时Flink解析器会把Kafka消息按你定义的schema映射成RowData。如果消息里多了一个字段默认会报错或者忽略如果字段类型不匹配比如定义成INT实际来的是字符串作业会直接脏数据。这个模式下复杂JSON反而成了负担schema管理极其痛苦。用RAW Format时整条消息就是一个STRING字段我把完整的JSON字符串交给自定义UDF路径提取完全在函数内部处理。好处是三句话能说清上游加字段完全不用改作业解析逻辑在函数里字段类型不合法不会让作业崩函数内部做了兜底调试阶段想打印整条原始JSON也可以直接透传非常直观。所以整体思路就是RAW收数据UDF做解析PRINT做输出。这个链路简单、可控、易排查适合作为复杂JSON接入Flink的第一版方案。2. 环境准备与Kafka接入的配置细节2.1 项目依赖和版本对齐我用的环境是Flink 1.171.16、1.18也适用主要注意connector版本匹配。Maven项目里核心依赖长这样properties flink.version1.17.2/flink.version scala.binary.version2.12/scala.binary.version /properties dependencies !-- Flink Table API -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- Kafka Connector -- dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version${flink.version}/version /dependency !-- JSON解析库这里用fastjson2纯粹是JSONPath好用 -- dependency groupIdcom.alibaba.fastjson2/groupId artifactIdfastjson2/artifactId version2.0.43/version /dependency /dependencies版本上有几个坑值得提前说flink-table-api-java-bridge用providedscope打包时不要打进去否则跟运行时自带的Table API冲突大概率报NoSuchMethodError。Kafka connector的版本必须和Flink主版本对应比如Flink 1.17就用flink-connector-kafka:1.17.2不要自己乱拉别的版本。如果用SQL Client提交UDF打成fat jar放进lib目录或者用-C参数指定依赖路径fastjson2也必须在classpath里。2.2 Kafka源表RAW模式才是复杂JSON的正确姿势建源表时我推荐写成这样CREATE TABLE kafka_source ( raw_json STRING ) WITH ( connector kafka, topic user_behavior_topic, properties.bootstrap.servers localhost:9092, properties.group.id flink_json_debug_group, properties.auto.offset.reset earliest, format raw );重点在这个format raw。很多人习惯性写成format json然后发现parse失败。原因很简单JSON format要求你显式声明每个字段同时它会把Kafka消息当JSON对象解析而复杂JSON里某个字段是数字还是字符串Kafka消息本身并没有schema信息解析器只能按你的表结构猜。猜对了是运气猜错了作业抛异常。RAW模式就没有这个问题Kafka里消息字节流是什么样raw_json字段就是什么样永远不解析失败。这才是处理“不规整复杂JSON”的最稳入口。2.3 打印输出表PRINT连接器的配置要点打印目标表我用内置PRINT连接器CREATE TABLE print_sink ( user_id STRING, event_id STRING, sku_id STRING, os STRING, raw_snapshot STRING ) WITH ( connector print );PRINT连接器会把每一条结果以I[字段1, 字段2, ...]的形式写到TaskManager的标准输出里查日志就能看到。这里有个非常实用的建议调试阶段务必把作业并行度设为1否则多个subtask同时往stdout打数据你根本对不上行与行的顺序看着像乱序其实是并行输出交错。SET parallelism.default 1;如果用的是DataStream API可以这样设置执行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1);打印虽然是调试期最常用的手段但它对结果集没有汇聚能力数据量大时日志会刷得飞快。想要控制打印量可以在SQL里加LIMIT仅批/TableResult模式支持或者配合HOP窗口、TUMBLE窗口只跑一小段。后面问题排查章节我再细说。3. 自定义get_json_object函数的核心实现3.1 ScalarFunction的基本骨架Flink Table API里实现用户自定义函数最简单的是继承ScalarFunction重写eval方法。它的基本规则是一个UDF类里可以有多个eval重载方法的第一个参数默认代表函数的第一个入参。我完整的实现如下使用fastjson2内置的JSONPath支持import org.apache.flink.table.functions.ScalarFunction; import org.apache.flink.table.annotation.DataTypeHint; import org.apache.flink.table.annotation.InputGroup; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONPath; public class GetJsonObject extends ScalarFunction { DataTypeHint(value STRING, bridgedTo String.class) public String eval( DataTypeHint(inputGroup InputGroup.ANY) Object jsonField, String jsonPath) { if (jsonField null || jsonPath null) { return null; } String jsonStr; if (jsonField instanceof String) { jsonStr (String) jsonField; } else { jsonStr jsonField.toString(); } if (jsonStr.trim().isEmpty()) { return null; } try { Object result JSONPath.of(jsonPath).extract(JSON.parse(jsonStr)); if (result null) { return null; } return result instanceof String ? (String) result : JSON.toJSONString(result); } catch (Exception e) { return null; } } }这段代码有几个设计是刻意为之的逐个说明。3.2 为什么用JSONPath而不是手动split解析自己写JSON解析最自然的想法是先parse成JSONObject然后一层层get。比如JSON.parseObject(jsonStr).getJSONObject(event).getJSONObject(properties).getJSONArray(items).getJSONObject(0).getString(skuId)。写起来不复杂但问题在于路径一变就得改代码。真实业务里JSON结构可能有几十个字段路径每条路径写一长串getCoder很容易麻。JSONPath的好处是把“路径”和“解析逻辑”解耦。代码里只需要传$.event.properties.items[0].skuId这种字符串函数内部统一处理。因为fastjson2的JSONPath支持$.common.userId取子节点$.event.properties.items[0]数组按下标取元素$.event.properties.items[*].skuId取数组所有元素的skuId字段返回数组$.common.device取整个子对象核心逻辑就一行JSONPath.of(jsonPath).extract(JSON.parse(jsonStr))。路径不对就抛异常函数捕获后返回NULL。这已经把八成需求都覆盖了。3.3 参数类型为什么要设计成Object Stringeval方法的第一个参数类型我没写死成String而是用DataTypeHint(inputGroup InputGroup.ANY)标注成Object再用注解指定输出是STRING。原因是我在实验中发现Flink SQL调用UDF时如果传入字段本身是STRING类型没问题但有些场景下用户会用CAST(raw_json AS STRING)后的结果再传给函数或者字段在Table层已经被声明成其他类型此时如果eval只接受String可能因为类型不匹配导致注册或调用失败。统一收成Object内部做String转换兼容性最好。实际开发我也建议所有“解析类”UDF都这样设计别为了省事把入参类型收窄。3.4 返回子对象和数组的策略get_json_object在Spark里有个行为值得抄过来如果路径指向的是一个JSON对象或数组返回的是对象序列化后的字符串而不是报错。上面的代码就是照这个行为做的return result instanceof String ? (String) result : JSON.toJSONString(result);比如传入$.common.device返回的就是{os:android,brand:xiaomi}这个JSON字符串。调试时想确认某个子结构的完整内容这一招特别管用。下游如果还想继续提里面的字段可以再调一次get_json_object形成链式解析。3.5 容错解析失败永远返回NULL而不是抛异常流式作业最忌讳一个坏数据让整个任务崩掉。Kafka里偶尔混入一条非JSON字符串、字段类型莫名其妙、路径因为上游版本变更失效都是家常便饭。所以函数里所有解析逻辑都包在try-catch里异常捕获后直接返回NULL。这个策略导致的结果就是解析不出数据时你不会看到报错只会看到字段为NULL。排查时就需要靠日志和审计这也是后面问题章节要讲的点。生产上我建议在UDF之上做一个配套的“原始数据监控”比如记录raw_json为NULL的比例一旦飙升就去查上游。3.6 注册函数并测试Flink SQL里注册临时函数的方法很简单CREATE TEMPORARY FUNCTION get_json_object AS com.example.udf.GetJsonObject;如果是通过TableEnvironment注册TableEnvironment tEnv TableEnvironment.create(EnvironmentSettings.inStreamingMode()); tEnv.createTemporarySystemFunction(get_json_object, GetJsonObject.class);这里有个容易翻车的细节CREATE TEMPORARY SYSTEM FUNCTION创建的是系统级函数所有database通用CREATE TEMPORARY FUNCTION创建的是会话级。在SQL Client里用CREATE TEMPORARY FUNCTION就够了但要注意函数类名必须带全限定名且JAR已经在classpath中。注册完先别急着接Kafka我习惯先在SQL Client里跑一个最简单的验证SELECT get_json_object( {common:{userId:u_10001},items:[{skuId:s_1}]}, $.items[0].skuId );能正确输出s_1再接Kafka联调省得两头排查。4. 完整实战从Kafka复杂JSON到控制台打印4.1 先造一条有代表性的复杂JSON为了演示完整链路我们往Kafka topicuser_behavior_topic里生产一条跟开头类似的数据这里再贴一遍后面建表和分析都基于它{ common: { appId: mall_app, userId: u_10001, ts: 1713177600000, device: {os: android, brand: xiaomi} }, event: { eventId: order_submit, properties: { orderAmount: 99.8, items: [ {skuId: s_8899, num: 1, price: 99.8, attrs: {color: black, size: L}} ] } }, traceId: trace_xxx_123 }生产消息的时候可以用Kafka自带命令行工具也可以用任意客户端脚本。关键是消息的value是上面这串JSON的原始文本不要加多余的前后缀。4.2 建表、创建函数、写查询完整跑通的SQL脚本如下-- 1) 源表raw格式拿整条JSON CREATE TABLE kafka_source ( raw_json STRING ) WITH ( connector kafka, topic user_behavior_topic, properties.bootstrap.servers localhost:9092, properties.group.id flink_json_debug_group, properties.auto.offset.reset earliest, format raw ); -- 2) 注册自定义UDF CREATE TEMPORARY FUNCTION get_json_object AS com.example.udf.GetJsonObject; -- 3) 打印目标表 CREATE TABLE print_sink ( user_id STRING, event_id STRING, first_sku_id STRING, os STRING, device_json STRING ) WITH ( connector print ); -- 4) 提取字段并打印 INSERT INTO print_sink SELECT get_json_object(raw_json, $.common.userId) AS user_id, get_json_object(raw_json, $.event.eventId) AS event_id, get_json_object(raw_json, $.event.properties.items[0].skuId) AS first_sku_id, get_json_object(raw_json, $.common.device.os) AS os, get_json_object(raw_json, $.common.device) AS device_json FROM kafka_source;查询逻辑里我特意放了两个不同类型的提取一个是原子字段一个是整个子对象。前者返回像u_10001这种普通字符串后者返回{os:android,brand:xiaomi}这种JSON字符串。如果你调试时需要把整条JSON也打出来直接在SELECT里加上raw_json字段就好。4.3 运行结果与日志解读作业跑起来后在TaskManager日志或者stdout里可以看到类似这样的输出I[u_10001, order_submit, s_8899, android, {os:android,brand:xiaomi}]I表示这条是Insert操作后面括号里的值就是SELECT语句定义的五个字段顺序一一对应。如果某个字段路径写错比如把$.event.properties.items[1].skuId当成路径下标1不存在对应位置就是null其它字段正常输出。这一点可以反向验证UDF容错逻辑是否生效。实时流里PRINT输出是没有终点的只要Kafka里持续有数据stdout就会一直刷。做一次性验证时建议用kafka-console-producer手动生产几条消息然后启动作业观察几秒确认结果后停掉作业。如果非要用自动生成的数据量大一些最好加个窗口或者LIMIT限制输出量。4.4 并行度对打印结果的影响前文提过打印调试时并行度设为1最省心。假如并行度是4Kafka分区有多个每个subtask都会独立打印自己消费的数据日志里会出现多个I[穿插在一起结果看起来就像错乱。这不是数据错乱是打印并发造成的假象。如果你只是想验证解析逻辑而不是验证并行处理能力先把并行度设为1再做打印。等到真正上线再把并行度调大并把PRINT sink换成Kafka或目标存储。5. 常见问题与排查技巧实录5.1 解析结果全是NULL这是最高频的问题原因分布我整理成表格可能原因具体表现排查方法路径写错所有目标字段都为NULL在SQL Client里用一条固定JSON试路径字段名大小写不一致只有部分字段为NULL对照原始JSON逐字符核对字段名源表不是raw格式作业直接失败或字段解析错乱检查WITH里format是不是rawJSON字符串前后有空白偶发解析失败在UDF里trim后再parse上游消息不是合法JSON整行NULL单独打印raw_json检查要避免这个问题的最高效手段一定是先在离线环境里把UDF配合固定JSON测一遍。路径写错是最常见错误比如把JSON里的字段名userId写成userid或者下标从1开始而不是0。这些都是肉眼难查的坑靠SQL Client单测几秒钟就能暴露。5.2 JSON字段名包含特殊字符怎么办有些JSON的key不是标准字母数字比如user-id、a.b这种。用$.common.user-id这种路径JSONPath会把它当成user减id解析结果自然不对。fastjson2在这种场景下支持把key用双引号括起来get_json_object(raw_json, $.user-id)我印象中这在Spark的get_json_object里也类似。平时写路径时如果发现解析结果不对先想是不是key里有特殊字符。这个细节很容易被忽略一旦踩中会浪费很多时间。5.3 打印数据量太大刷爆日志PRINT连接器默认是每条数据都输出数据量大时TaskManager的日志文件会快速膨胀最终影响磁盘和排查。调试阶段我通常这样控制在Kafka生产端只发几条测试消息或者源表加上scan.startup.mode earliest-offset并且用一次性运行跑完取消作业或者SQL里构造窗口比如只统计最近一小段窗口内的数据再打印严格意义上PRINT连接器没有“限流”参数控制输出量要靠作业本身。如果你只是临时验证最快的方式是启动作业前先用命令行工具确认topic里数据条数再决定要不要跑。5.4 UDF注册不生效常见现象是SQL Client执行CREATE TEMPORARY FUNCTION时报ClassNotFoundException或者作业启动时函数找不到。通常是JAR没打进去或者类名拼错。用SQL Client时启动命令要带上JARflink sql-client -j flink-udf.jar如果用-C参数指定目录要确认目录下真的有这个jar。注意SQL Client里的函数是会话级的重启SQL Client后需要重新注册。生产环境建议用Catalog持久化函数或者直接在作业代码里注册避免每次手动操作。5.5 fastjson2和其他JSON库的依赖冲突Flink本身带了一些Jackson相关依赖UDF里用fastjson2一般没事但如果项目中同时引入了多个JSON库可能出现版本冲突导致奇奇怪怪的NoSuchMethodError。我的做法是UDF所在的JAR使用shade插件打包把fastjson2重定位relocation到自定义包名避免跟运行时依赖冲突。下面是一个简化版shade配置plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration relocations relocation patterncom.alibaba.fastjson2/pattern shadedPatterncom.example.shaded.fastjson2/shadedPattern /relocation /relocations filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters /configuration /execution /executions /plugin5.6 性能与吞吐JSONPath解析会不会成为瓶颈有人会担心每条数据都做一次JSON parse和JSONPath extract吞吐上不去。实测下来单条消息几百字节的规模单并行度每秒处理几千条完全没问题。真正需要高性能的场景可以从两个方向优化复用JSONPath实例不要每次都of(jsonPath)提前编译好常用路径对象传入的字符串如果可以直接让上游规范成紧凑JSON减少序列化开销。但说句实在话对于绝大多数日志解析、埋点清洗场景自定义UDF的解析性能不是瓶颈Kafka的消费能力和下游存储往往才是。6. 个人实操经验工具箱越简单越可靠整个项目做完我个人最深的体会是处理Flink里的复杂JSON不要一上来就整复杂的schema推导或类型映射。先用RAW模式接住字符串再配合一个灵活的JSONPath提取函数是最快、最稳、最好排查的路径。你可能最后发现这个UDF占了80%的解析工作后面就算要接正经的schema映射也可以在这个函数返回结果的基础上去做。另外一个小技巧给这个UDF加一个“原始数据透传”重载方法。比如eval(String json, String path)和eval(String json)后者直接返回原始JSON字符串。调试的时候把raw_json和解析字段一起打印出问题一眼就能定位是解析逻辑的问题还是上游数据格式的问题。这个自定义函数的模式还可以继续扩展。比如把返回的JSON字符串再做一次CAST AS或JSON_QUERY操作就能满足更复杂的后续加工或者把这个函数放进持久化Catalog团队其他人直接SELECT get_json_object(x, $.y)就能用不用每个人都重写一遍。处理Kafka复杂JSON这件事本质上不是Flink能力不够而是你需要一个趁手的工具。get_json_object就是那个工具花半小时写完后面能省无数个小时。