Java 类名：com.alibaba.alink.operator.stream.dataproc.JsonValueStreamOp
Python 类名：JsonValueStreamOp

功能介绍

该组件完成json字符串中的信息抽取，按照用户给定的Path 抓取出相应的信息。该组件支持

  • 按照多JsonPath规则编写的多条抽取规则。
  • 默认输出类型为string，也可以根据实际情况指定数据类型

    JsonPath表达式

    | 元素 | 含义 | | —- | —- |

| $ | 表示文档的根元素 |

| @ | 表示文档的当前元素 |

| .node_name 或 [‘node_name’] | 匹配下级节点 |

| [index] | 选择数组中的元素 |

| [start🔚step] | 表示数组切片语法 |

  • 注意事项：jsonPath参数是字符串数组，长度应当和指定的输出列名称数目保持一致。

    参数说明

    | 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| jsonPath | Json 路径数组 | 用来指定 Json 抽取的内容。 | String[] | ✓ | | |

| outputCols | 输出结果列列名数组 | 输出结果列列名数组，必选 | String[] | ✓ | | |

| selectedCol | 选中的列名 | 计算列对应的列名 | String | ✓ | 所选列类型为 [STRING] | |

| outputColTypes | 输出结果列列类型数组 | 输出结果列类型数组 | String[] | | | null |

| reservedCols | 算法保留列名 | 算法保留列 | String[] | | | null |

| skipFailed | 是否跳过错误 | 当遇到抽取值为null 时是否跳过 | boolean | | | false |

| numThreads | 组件多线程线程个数 | 组件多线程线程个数 | Integer | | | 1 |

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. df = pd.DataFrame([
  8.      ["{a:boy,b:{b1:[1,2],b2:2}}"],
  9.      ["{a:girl,b:{b1:[1,3],b2:2}}"]
  10.  ])
  12. streamData = StreamOperator.fromDataframe(df, schemaStr='str string')
  14. JsonValueStreamOp()\
  15.      .setJsonPath(["$.a", "$.b.b1[0]","$.b.b2"])\
  16.      .setSelectedCol("str")\
  17.      .setOutputCols(["f0","f1","f2"])\
  18.      .setOutputColTypes(["string", "int", "long"])\
  19.      .linkFrom(streamData)\
  20.      .print()
  22. StreamOperator.execute()

Java 代码

  1. import org.apache.flink.types.Row;
  3. import com.alibaba.alink.operator.stream.StreamOperator;
  4. import com.alibaba.alink.operator.stream.dataproc.JsonValueStreamOp;
  5. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  6. import org.junit.Test;
  8. import java.util.Arrays;
  9. import java.util.List;
  11. public class JsonValueStreamOpTest {
  12.     @Test
  13.     public void testJsonValueStreamOp() throws Exception {
  14.         List <Row> df = Arrays.asList(
  15.             Row.of("{a:boy,b:{b1:[1,2],b2:2}}"),
  16.             Row.of("{a:girl,b:{b1:[1,3],b2:2}}")
  17.         );
  18.         StreamOperator <?> streamData = new MemSourceStreamOp(df, "str string");
  19.         new JsonValueStreamOp()
  20.             .setJsonPath("$.a", "$.b.b1[0]","$.b.b2")
  21.             .setSelectedCol("str")
  22.             .setOutputCols("f0", "f1", "f2")
  23.             .setOutputColTypes("string", "int", "long")
  24.             .linkFrom(streamData)
  25.             .print();
  26.         StreamOperator.execute();
  27.     }
  28. }

运行结果

| str | f0 | f1 | f2 | | —- | —- | —- | —- |

| {a:girl,b:{b1:[1,3],b2:2}} | girl | 1 | 2 |

| {a:boy,b:{b1:[1,2],b2:2}} | boy | 1 | 2 |