JSON值抽取 (JsonValueBatchOp)

Java 类名：com.alibaba.alink.operator.batch.dataproc.JsonValueBatchOp

Python 类名：JsonValueBatchOp

功能介绍

该组件完成json字符串中的信息抽取，按照用户给定的Path 抓取出相应的信息。该组件支持
- 按照多JsonPath规则编写的多条抽取规则。
- 指定输出列的数据类型

JsonPath表达式

元素	含义
$	表示文档的根元素
@	表示文档的当前元素
.node_name 或 [‘node_name’]	匹配下级节点
[index]	选择数组中的元素
[ start : end : 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 代码

from pyalink.alink import *

import pandas as pd

useLocalEnv(1)

df = pd.DataFrame([
     ["{a:boy,b:{b1:[1,2],b2:2}}"],
     ["{a:girl,b:{b1:[1,3],b2:2}}"]
 ])
 
data = BatchOperator.fromDataframe(df, schemaStr='str string')
 
JsonValueBatchOp()\
     .setJsonPath(["$.a", "$.b.b1[0]","$.b.b2"])\
     .setSelectedCol("str")\
     .setOutputCols(["f0","f1","f2"])\
     .linkFrom(data)\
     .print()

Java 代码

import org.apache.flink.types.Row;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.JsonValueBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class JsonValueStreamOpTest {
	@Test
	public void testJsonValueBatchOp() throws Exception {
		List <Row> df = Arrays.asList(
			Row.of("{a:boy,b:{b1:[1,2],b2:2}}"),
			Row.of("{a:girl,b:{b1:[1,3],b2:2}}")
		);
		BatchOperator <?> data = new MemSourceBatchOp(df, "str string");
		new JsonValueBatchOp()
			.setJsonPath("$.a", "$.b.b1[0]","$.b.b2")
			.setSelectedCol("str")
			.setOutputCols("f0", "f1","f2")
			.linkFrom(data)
			.print();
	}
}

运行结果

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