数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载JsonPath transform plugin —— 支持使用 JSONPath 语法从上游字段中按路径选择数据并将结果以指定类型输出为新的目标字段。本文是 SeaTunnel 中 JsonPath 转换插件的实战指南。它适用于这样一类场景上游 Source如 Kafka、JDBC、FakeSource产出的数据中某个字段承载着嵌套 JSON 字符串或结构化对象而你希望在进入 Sink 之前仅用一行 JSONPath 表达式就把其中的深层字段精准抽取出来并完成类型转换与字段重命名。读完本文你将掌握 JsonPath 插件的全部配置项、两种典型使用模式解析 JSON 字段、解析 SeatunnelRow 字段、底层实现原理以及常见错误码的排查方法可以直接复制示例配置投入生产。插件定位一个多字段输出的字段级转换器JsonPath 转换插件在 SeaTunnel 中注册名为JsonPath与Sql、Split、Replace等插件一样位于 transform 阶段承担 Source 与 Sink 之间的中间处理。从源码结构看它的核心实现位于 JsonPathTransform.java插件声明PLUGIN_NAME JsonPath通过 JsonPathTransformFactory.java 以AutoService(Factory.class)方式注册工厂标识factoryIdentifier即为JsonPath。在类继承关系上JsonPathTransform继承自MultipleFieldOutputTransform见 MultipleFieldOutputTransform.java这意味着它一次可以产出多个新字段而不仅仅是单字段。其 schema 变换逻辑如下若输出的dest_field与输入表中已有字段同名则复用该列仅当其数据类型不一致时用输出类型覆盖原列类型若dest_field是全新字段名则在原表 schema 末尾追加新列追加新列时会通过SeaTunnelRowContainerGenerator生成扩展容量的新行容器原字段按原位置拷贝原表的tableId与rowKind一并保留。也就是说JsonPath 转换属于增强型字段变换它不会丢弃源表原有列而是在保留全部原始字段的同时把抽取出的目标字段作为新列追加或覆盖同名列输出。配置总览JsonPath 插件只有一个顶层配置项columns且为必填nametyperequireddefault valueColumnsArrayYes公共选项与其他转换插件一致JsonPath 同样支持source_table_name与result_table_name两个公共参数用于在 transform 链中衔接数据表source_table_name指定当前插件处理哪个上游数据集。不指定时处理配置文件中前一个插件输出的数据result_table_name将处理后结果注册为可被其他插件通过source_table_name直接引用的临时表。详细说明可参考 Transform 公共选项文档。配置示例中的source_table_name fake、result_table_name fake1即表示读取名为fake的表转换后注册为fake1供下游 Sink 使用。columns 数组中的字段定义columns是一个数组每个元素代表一条抽取规则包含以下子参数nametyperequireddefault valuesrc_fieldStringYesdest_fieldStringYespathStringYesdest_typeStringNoStringsrc_field —— JSON 源字段要解析的源字段名。该字段的值承载 JSON 内容支持以下 SeatunnelDataType见 JsonPathTransform.java 中的doTransform分支处理STRING直接以字符串内容作为 JSON 文本解析BYTES按new String((byte[]) value)解码为字符串后再解析ARRAY/MAP通过JsonUtils.toJsonString(value)序列化为 JSON 文本后解析ROW对SeaTunnelRow类型的嵌套行将其getFields()序列化为 JSON 文本后解析。除上述五种类型之外的其他类型会抛出unsupportedDataType错误源码位于doTransform的default分支。注意插件在初始化时initSrcFieldIndexArr会校验src_field必须真实存在于输入表的字段列表中否则直接抛出cannotFindInputFieldError错误码对应JSONPATH_ERROR_CODE-02JsonPathTransform src_field not found in source。dest_field —— 输出字段名使用 JSONPath 抽取并转换后结果将写入的目标字段名。若与源表已有列重名则覆盖该列类型否则作为新列追加到表末尾。dest_type —— 输出字段类型目标字段的数据类型默认值为string。该参数通过SeaTunnelDataTypeConvertorUtil.deserializeSeaTunnelDataType(srcField, type)解析见 JsonPathTransformConfig.java因此支持所有 SeaTunnel 数据类型写法例如标量类型string、boolean、int、float、double、date、time、timestamp带精度类型decimal(4,2)复杂类型arraystring、mapstring,string、嵌套row(...)等底层转换由JsonToRowConverters根据dest_type生成对应的JsonToObjectConverterJsonPathTransform.javaJSON 抽取结果先转为JsonNode再由该 converter 完成 JSON 节点到 SeaTunnel 类型的转换。path —— JSONPath 表达式要执行的 JSONPath 路径如$.data.c_string、$[0]、$.data.c_array[0]等。插件使用 Jayway 的com.jayway.jsonpath.JsonPath进行编译与读取并借助ConcurrentHashMap做全局编译缓存相同路径只编译一次后续所有行直接复用编译结果JsonPathTransform.java在高吞吐流式场景下可显著降低重复编译开销。读取 JSON 示例假设上游 Source 产出的每一行数据中字段data的值为如下 JSON{ data: { c_string: this is a string, c_boolean: true, c_integer: 42, c_float: 3.14, c_double: 3.14, c_decimal: 10.55, c_date: 2023-10-29, c_datetime: 16:12:43.459, c_array:[item1, item2, item3] } }我们希望用 JSONPath 一次性抽取其中的全部属性并分别指定目标字段名与目标类型transform { JsonPath { source_table_name fake result_table_name fake1 columns [ { src_field data path $.data.c_string dest_field c1_string }, { src_field data path $.data.c_boolean dest_field c1_boolean dest_type boolean }, { src_field data path $.data.c_integer dest_field c1_integer dest_type int }, { src_field data path $.data.c_float dest_field c1_float dest_type float }, { src_field data path $.data.c_double dest_field c1_double dest_type double }, { src_field data path $.data.c_decimal dest_field c1_decimal dest_type decimal(4,2) }, { src_field data path $.data.c_date dest_field c1_date dest_type date }, { src_field data path $.data.c_datetime dest_field c1_datetime dest_type time }, { src_field data path $.data.c_array dest_field c1_array dest_type arraystring } ] } }转换后结果表fake1中除保留原始data列外会新增 9 个目标列形如datac1_stringc1_booleanc1_integerc1_floatc1_doublec1_decimalc1_datec1_datetimec1_arraytoo much content not to showthis is a stringtrue423.143.1410.552023-10-2916:12:43.459[item1, item2, item3]从示例可以看到几个实用要点未显式配置dest_type的第一条规则默认输出string每条规则可以混合使用不同目标类型互不影响decimal(4,2)这类带精度写法、arraystring这类复杂类型写法均被支持。读取 SeatunnelRow 示例除了 JSON 字符串src_field还可以是ROWSeatunnelRow类型——例如上游通过 SQL 或嵌套 schema 构造出的结构体列。假设数据行中有一个名为col的列类型为 SeatunnelRow其内部结构如下SeatunnelRow(col)othernameage....a18....也就是说col的字段依次为name值为a与age值为18。JsonPath 转换会先把 SeatunnelRow 的字段值序列化为一个 JSON 数组[a, 18]因此可以用数组下标形式的 JSONPath 进行抽取transform { JsonPath { source_table_name fake result_table_name fake1 columns [ { src_field col path $[0] dest_field name dest_type string }, { src_field col path $[1] dest_field age dest_type int } ] } }转换后结果表fake1将新增name与age两个字段nameagecolothera18[a,18]...该场景背后对应源码中doTransform的ROW分支jsonString JsonUtils.toJsonString(row.getFields())即把SeaTunnelRow的字段数组整体序列化为 JSON 数组文本后再交给 JSONPath 读取JsonPathTransform.java。这意味着抽取嵌套 ROW 时下标$[0]、$[1]与行内字段的声明顺序一一对应。配置校验与错误码JsonPath 插件的配置在 JsonPathTransformConfig.java 的of()方法中完成解析与校验错误码定义见 JsonPathTransformErrorCode.java错误码触发条件JSONPATH_ERROR_CODE-01未配置columnsCOLUMNS_MUST_NOT_EMPTYJSONPATH_ERROR_CODE-02src_field为空或src_field在源表中不存在JSONPATH_ERROR_CODE-03path为空PATH_MUST_NOT_EMPTYJSONPATH_ERROR_CODE-04dest_field为空DEST_FIELD_MUST_NOT_EMPTYJSONPATH_ERROR_CODE-05JSONPath 表达式非法编译或读取失败JSON_PATH_COMPILE_ERRORJSONPATH_ERROR_CODE-06dest_type为空DEST_TYPE_MUST_NOT_EMPTY其中校验规则checkColumnConfig对每一条抽取规则逐一检查path、src_field、dest_field三者均非空白columns缺失则在插件初始化阶段直接抛错因此配置错误会在作业启动时而非运行期暴露。JSONPATH_ERROR_CODE-05则发生在运行期当某一行数据的 JSONPath 读取抛出JsonPathException时会以TransformException形式向上抛出并携带原始错误信息JsonPathTransform.java便于定位具体是哪条路径写错。对应单元测试可参考 JsonPathTransformFactoryTest.java其验证了工厂optionRule()要求columns为必填可正常构建。使用建议与注意事项配置一次、多处复用同一src_field可以出现在多条抽取规则中如示例中 9 条规则全部指向data插件会按规则顺序逐条执行抽取互不干扰。路径编译缓存相同的path全局只编译一次JSON_PATH_CACHE为ConcurrentHashMap生产环境中应尽量让规则中的 path 保持稳定、可复用以发挥缓存收益。dest_type 与 path 结果要匹配path抽取出的 JSON 节点会被JsonToRowConverters按dest_type强制转换例如对字符串字段指定int可能触发转换失败配置时应保证路径指向的节点类型与目标类型一致。ROW 列的下标顺序使用$[0]、$[1]抽取 SeatunnelRow 时下标顺序即行内字段声明顺序字段增减会导致下标错位需同步调整配置。空值安全当src_field对应值为null时doTransform直接返回null而不执行 JSONPathJsonPathTransform.java下游需自行处理空值。更新日志新增 JsonPath Transform 插件支持通过 JSONPath 从 STRING、BYTES、ARRAY、MAP、ROW 类型字段中按路径抽取数据并支持dest_type类型转换与多字段批量输出。赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐SeaTunnel JsonPath 转换插件详解用 JSONPath 从 JSON 字段中精准抽取数据SeaTunnel JsonPath 转换插件详解用 JSONPath 从 JSON 字段中精准抽取数据 本文基于 Apache SeaTunnelSeaT数据工程大数据批处理流处理SeaTunnel JsonPath 转换插件实战指南用 JSONPath 从 JSON 数据中高效提取字段SeaTunnel JsonPath 转换插件实战指南用 JSONPath 从 JSON 数据中高效提取字段 导读 本文围绕 SeaTunnel 的 Json数据集成ETL大数据批处理流处理变更数据捕获Apache SeaTunnel中的JSONPath转换插件详解Apache SeaTunnel中的JSONPath转换插件详解 什么是JSONPath转换插件 JSONPath转换插件是Apache SeaTunnel数据数据集成ETL大数据批处理流处理变更数据捕获上一篇LeetCode-Go 题解1034. Coloring A Border连通分量边界着色DFS 实现全解析下一篇Vant Empty 空状态组件实战指南占位提示、内置图片类型与主题定制创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考