[Feature][Transform-V2] Support jsonpath extracting multiple fields in a configuration block (#9712)

xiaochen-zhou · web-flow · commit c57bdd19c05c · 2025-08-26T17:57:02.000+08:00
diff --git a/docs/en/transform-v2/jsonpath.md b/docs/en/transform-v2/jsonpath.md
@@ -161,6 +161,27 @@ transform {
 }
 ```
 
+The same result can be achieved with much simpler configuration using batch field extraction with array format:
+
+```hocon
+transform {
+  JsonPath {
+    plugin_input = "fake"
+    plugin_output = "fake1"
+    columns = [
+     {
+        "src_field" = "data"
+        "path" = ["$.data.c_string", "$.data.c_boolean", "$.data.c_integer", "$.data.c_float", "$.data.c_double", "$.data.c_decimal", "$.data.c_date", "$.data.c_datetime", "$.data.c_array", "$.data.c_map_array"]
+        "dest_field" = ["c1_string", "c1_boolean", "c1_integer", "c1_float", "c1_double", "c1_decimal", "c1_date", "c1_datetime", "c1_array", "c1_map_array"]
+        "dest_type" = ["string", "boolean", "int", "float", "double", "decimal(4,2)", "date", "time", "array<string>", "array<map<string, string>>"]
+     }
+    ]
+  }
+}
+```
+
+**Important:** When using batch field extraction (multiple paths, dest_fields, and dest_types), the `dest_type` parameter is **required** and cannot be omitted. Each extracted field must have a corresponding type specified. The array format provides better readability and is less error-prone than string-based configurations.
+
 Then the data result table `fake1` will like this
 
 |             data             |    c1_string     | c1_boolean | c1_integer | c1_float | c1_double | c1_decimal |  c1_date   | c1_datetime  |          c1_array           |
diff --git a/docs/zh/transform-v2/jsonpath.md b/docs/zh/transform-v2/jsonpath.md
@@ -161,6 +161,26 @@ transform {
 }
 ```
 
+使用批量字段提取功能可以用更简洁的数组格式配置实现相同的结果：
+
+```hocon
+transform {
+  JsonPath {
+    plugin_input = "fake"
+    plugin_output = "fake1"
+    columns = [
+     {
+        "src_field" = "data"
+        "path" = ["$.data.c_string", "$.data.c_boolean", "$.data.c_integer", "$.data.c_float", "$.data.c_double", "$.data.c_decimal", "$.data.c_date", "$.data.c_datetime", "$.data.c_array", "$.data.c_map_array"]
+        "dest_field" = ["c1_string", "c1_boolean", "c1_integer", "c1_float", "c1_double", "c1_decimal", "c1_date", "c1_datetime", "c1_array", "c1_map_array"]
+        "dest_type" = ["string", "boolean", "int", "float", "double", "decimal(4,2)", "date", "time", "array<string>", "array<map<string, string>>"]
+     }
+    ]
+  }
+}
+```
+**重要提示：** 当使用批量字段提取（多个 paths、dest_fields 和 dest_types）时，`dest_type` 参数是必填的，不能省略。每个提取的字段都必须指定一个对应的类型。数组格式提供了更好的可读性，比基于字符串的配置更不容易出错。
+
 那么数据结果表 `fake1` 将会像这样
 
 |             data             |    c1_string     | c1_boolean | c1_integer | c1_float | c1_double | c1_decimal |  c1_date   | c1_datetime  |          c1_array           |
@@ -210,6 +230,8 @@ transform {
 |------|-----|----------|-------|
 | a    | 18  | ["a",18] | ...   |
 
+
+
 ## 配置异常数据处理策略
 
 您可以配置 `row_error_handle_way` 与 `column_error_handle_way` 来处理异常数据，两者都是非必填项。
diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestJsonPathTransformIT.java b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestJsonPathTransformIT.java
@@ -66,4 +66,11 @@ public void testArrayType(TestContainer container) throws Exception {
                 container.executeJob("/json_path_transform/json_path_array_map.conf");
         Assertions.assertEquals(0, execResult.getExitCode());
     }
+
+    @TestTemplate
+    public void testBatchFields(TestContainer container) throws Exception {
+        Container.ExecResult execResult =
+                container.executeJob("/json_path_transform/json_path_batch_fields_test.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
 }
diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/json_path_transform/json_path_batch_fields_test.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/json_path_transform/json_path_batch_fields_test.conf
@@ -0,0 +1,269 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file tests JsonPath batch fields extraction functionality
+######
+
+env {
+  job.mode = "BATCH"
+}
+
+source {
+  FakeSource {
+    plugin_output = "fake"
+    row.num = 100
+    string.fake.mode = "template"
+    string.template=["{\"mysql_fields\":{\"id\":{\"v\":1001},\"code\":{\"v\":\"TEST001\"},\"group_code\":{\"v\":\"GROUP001\"},\"user_id\":{\"v\":2001},\"patient_id\":{\"v\":3001},\"doctor_id\":{\"v\":4001},\"price\":{\"v\":99.99},\"status\":{\"v\":1},\"create_time\":{\"v\":\"2023-10-29 10:30:00\"},\"update_time\":{\"v\":\"2023-10-29 11:30:00\"}},\"nested_data\":{\"user\":{\"profile\":{\"name\":\"John\",\"age\":30},\"settings\":{\"theme\":\"dark\",\"lang\":\"en\"}},\"orders\":[{\"id\":101,\"amount\":50.5},{\"id\":102,\"amount\":75.8}]},\"array_fields\":[{\"type\":\"A\",\"value\":100},{\"type\":\"B\",\"value\":200}]}"]
+    schema = {
+      fields {
+        mysql_fields = "string"
+      }
+    }
+  }
+}
+
+transform {
+  JsonPath {
+    plugin_input = "fake"
+    plugin_output = "fake1"
+    columns = [
+     {
+        "src_field" = "mysql_fields"
+        "path" = ["$.mysql_fields.id.v", "$.mysql_fields.code.v", "$.mysql_fields.group_code.v", "$.mysql_fields.user_id.v", "$.mysql_fields.patient_id.v", "$.mysql_fields.doctor_id.v", "$.mysql_fields.price.v", "$.mysql_fields.status.v", "$.mysql_fields.create_time.v", "$.mysql_fields.update_time.v", "$.nested_data.user.profile.name", "$.nested_data.user.profile.age", "$.nested_data.user.settings.theme", "$.nested_data.orders[0].id", "$.nested_data.orders[0].amount", "$.nested_data.orders[1].id", "$.array_fields[0].type", "$.array_fields[0].value", "$.array_fields[1].type", "$.array_fields[1].value"]
+        "dest_field" = ["id", "code", "group_code", "user_id", "patient_id", "doctor_id", "price", "status", "create_time", "update_time", "user_name", "user_age", "user_theme", "first_order_id", "first_order_amount", "second_order_id", "first_type", "first_value", "second_type", "second_value"]
+        "dest_type" = ["bigint", "string", "string", "bigint", "bigint", "bigint", "double", "int", "string", "string", "string", "int", "string", "int", "double", "int", "string", "int", "string", "int"]
+     }
+    ]
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = "fake1"
+    rules =
+      {
+        row_rules = [
+          {
+            rule_type = MIN_ROW
+            rule_value = 100
+          }
+        ],
+        field_rules = [
+          {
+            field_name = id
+            field_type = bigint
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 1001
+              }
+            ]
+          },
+          {
+            field_name = code
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "TEST001"
+              }
+            ]
+          },
+          {
+            field_name = group_code
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "GROUP001"
+              }
+            ]
+          },
+          {
+            field_name = user_id
+            field_type = bigint
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 2001
+              }
+            ]
+          },
+          {
+            field_name = patient_id
+            field_type = bigint
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 3001
+              }
+            ]
+          },
+          {
+            field_name = doctor_id
+            field_type = bigint
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 4001
+              }
+            ]
+          },
+          {
+            field_name = price
+            field_type = double
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 99.99
+              }
+            ]
+          },
+          {
+            field_name = status
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 1
+              }
+            ]
+          },
+          {
+            field_name = create_time
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "2023-10-29 10:30:00"
+              }
+            ]
+          },
+          {
+            field_name = update_time
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "2023-10-29 11:30:00"
+              }
+            ]
+          },
+          {
+            field_name = user_name
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "John"
+              }
+            ]
+          },
+          {
+            field_name = user_age
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 30
+              }
+            ]
+          },
+          {
+            field_name = user_theme
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "dark"
+              }
+            ]
+          },
+          {
+            field_name = first_order_id
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 101
+              }
+            ]
+          },
+          {
+            field_name = first_order_amount
+            field_type = double
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 50.5
+              }
+            ]
+          },
+          {
+            field_name = second_order_id
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 102
+              }
+            ]
+          },
+          {
+            field_name = first_type
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "A"
+              }
+            ]
+          },
+          {
+            field_name = first_value
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 100
+              }
+            ]
+          },
+          {
+            field_name = second_type
+            field_type = string
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = "B"
+              }
+            ]
+          },
+          {
+            field_name = second_value
+            field_type = int
+            field_value = [
+              {
+                rule_type = NOT_NULL
+                equals_to = 200
+              }
+            ]
+          }
+        ]
+      }
+  }
+}
diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformConfig.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/jsonpath/JsonPathTransformConfig.java
diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/JsonPathTransformTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/JsonPathTransformTest.java

Original file line number	Diff line number	Diff line change
`@@ -66,4 +66,11 @@ public void testArrayType(TestContainer container) throws Exception {`
`66`	`66`	`container.executeJob("/json_path_transform/json_path_array_map.conf");`
`67`	`67`	`Assertions.assertEquals(0, execResult.getExitCode());`
`68`	`68`	`}`
	`69`	`+`
	`70`	`+ @TestTemplate`
	`71`	`+ public void testBatchFields(TestContainer container) throws Exception {`
	`72`	`+ Container.ExecResult execResult =`
	`73`	`+ container.executeJob("/json_path_transform/json_path_batch_fields_test.conf");`
	`74`	`+ Assertions.assertEquals(0, execResult.getExitCode());`
	`75`	`+ }`
`69`	`76`	`}`