EncoderDecoder

该方法将输入的表数据 obj 进行批量 解析、转换 处理，并输出到创建 coder 时传入的输出 handler 中。

如果该 coder 未指定输出 handler，则无法执行该函数，执行时会报错 To use method parseAndHandle, the handler of the coder object must be specified.

语法

coder.parseAndHandle(obj)

参数

• obj：具体输入参数要求见各类 coder 的说明，通常输入参数可以为 scalar，vector 或者 内存表。

返回值

布尔类型

解析、转换 成功返回 true，得到的 table 会输出到 coder 的输出端。解析失败返回 false，输出相关解析报错信息到日志

该方法将输入的表数据 obj 进行批量 解析/编码 处理，直接返回结果。

语法

coder.parse(obj)

参数

• obj：具体输入参数要求见各类 coder 的说明，通常输入参数可以为 scalar，vector 或者 内存表。

返回值

内存表

解析成功返回一个内存表，解析失败会输出具体错误信息到日志，或者抛出异常，视不同的 coder 而定。

语法

EncoderDecoder::jsonDecoder(colNames, colTypes, [handler], [workerNum=1], [batchSize=0], [throttle=1], [isMultiJson=false])

参数

• colNames：指定要解析的 json 数据中的 key 的名称。它是 json 中的键，类型为 string 类型的 vector。

• colTypes: 指定要解析的 json 数据中的 value 的类型。它是 json 中的值类型，类型为 int 类型的 vector。

• handler：可以是一个表，用于输出解析结果，也可以是一个函数对象，当前 decoder 的输出将作为该函数的输入进行处理。

• workerNum：处理解析运算的工作线程数量。默认值是 1.

• batchSize：是一个整数。若为正数，表示未处理消息的数量达到 batchSize 时，handler 才会处理消息。若未指定或为非正数，每一批次的消息到达之后，handler 就会马上处理。默认值是 0。

• throttle：是一个浮点数，单位为秒，默认值为 1。表示继上次 handler 处理消息之后，若 batchSize 条件一直未达到，多久后再次处理消息。

• isMultiJson：是一个布尔值，默认为 false. 表示一行内是否存在多个 JSON 字符串。如果设置为 true, 会先进行字符串分割再进行解析，需要注意，设置为 true 时要求 json 内字符串格式的 value 不能带字符 } 和 {

返回值

解析 json 数据的 coder 对象

jsonDecoder 对象函数 parseAndHandle

parseAndHandle 方法解析 json 数据，输出到 handler 中

方法的输入参数可以为 scalar，vector 或者 内存表。如果是 scalar 和 vector 类型则必须为 string 类型，如果是内存表，则必须只有一列，且列类型为 string 类型

如果 schema 不匹配，抛出异常 Usage: coder.parseAndHandle(obj). Obj must be a table with one STRING column。

如果其中的 json 数据格式非法，则会在 log 中输出 syntax error while parsing value (具体异常信息) 相关行，但仍然会输出字段全为空的结果到目标表中。

jsonDecoder 对象函数 parse

parse 方法解析输入的数据，输入参数类型同 parseAndHandle。返回 table。如果 json 输入非法，会输出字段全为空的结果。如果某个字段不出现，则会被自动赋空值。

数据类型与转换

colTypes 支持填写 BOOL、CHAR、INT、LONG、FLOAT、DOUBLE、STRING、BLOB 以及 arrayVector 类型 BOOL[]、INT[]、LONG[]、FLOAT[]、DOUBLE[]。

转换规则

不在表格中出现的类型转换均为不支持

精度说明：超出 double、long 等上限的数据，可以转为 STRING 类型保留精度。

示例

创建数据

str={"int": -637811349, "long": 637811349473772538,"float":-0.004 ,"double": 12133.938,"string": "2022-02-22 13:55:47.377"}
data = [str.toStdJson(),str.toStdJson(),str.toStdJson(),str.toStdJson()]
appendData = table(data as `string)

使用未指定输出的 jsonDecoder 调用 parse 进行解析

colNames = ["int", "long","float","double","string"]
colTypes = [INT, LONG, FLOAT , DOUBLE, STRING]
coder1 = EncoderDecoder::jsonDecoder(colNames, colTypes)
coder1.parse(appendData)

输出如下

int        long               float  double                string
---------- ------------------ ------ --------------------- -----------------------
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377

使用指定输出及其他参数的 jsonDecoder 调用 parseAndHandle 进行解析

handler = table(1:0, colNames, colTypes)
coder2 = EncoderDecoder::jsonDecoder(colNames, colTypes, handler, 4, 100, 0.01)
coder2.parseAndHandle(appendData)

handler 内容如下

int        long               float  double                string
---------- ------------------ ------ --------------------- -----------------------
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377
-637811349 637811349473772538 -0.004 12133.938000000000101 2022-02-22 13:55:47.377

使用指定 handler 为函数的 jsonDecoder 调用 parseAndHandle 进行解析

colNames2 = ["int", "long","float","double","date", "time"]
colTypes2 = [INT, LONG, FLOAT , DOUBLE, TIMESTAMP, TIMESTAMP]
dest = table(1:0, colNames2, colTypes2)

def parserDef(msg, mutable dest) {
    t = table(msg[`int] as `int, msg[`long] as `long, msg[`float] as `float, msg[`double] as `double, temporalParse(msg[`string] ,"yyyy-MM-dd HH:mm:ss.SSS")  as `date);
    t.update!(`name, now());
    dest.append!(t);
}
coder3 = EncoderDecoder::jsonDecoder(colNames, colTypes, parserDef{, dest}, 4, 100, 1)
coder3.parseAndHandle(appendData)

dest 表格内容如下

int        long               float  double                date                    time
---------- ------------------ ------ --------------------- ----------------------- -----------------------
-637811349 637811349473772538 -0.004 12133.938000000000101 2022.02.22T13:55:47.377 2023.10.26T11:22:00.140
-637811349 637811349473772538 -0.004 12133.938000000000101 2022.02.22T13:55:47.377 2023.10.26T11:22:00.140
-637811349 637811349473772538 -0.004 12133.938000000000101 2022.02.22T13:55:47.377 2023.10.26T11:22:00.140
-637811349 637811349473772538 -0.004 12133.938000000000101 2022.02.22T13:55:47.377 2023.10.26T11:22:00.140

// 用法1
coder = EncoderDecoder::protobufDecoder("path/to/proto");
coder.parse(data);

// 用法2
schema = EncoderDecoder::extractProtobufSchema("path/to/proto");
outputTable = table(1:0, schema[`name], schema[`type]);
coder = EncoderDecoder::protobufDecoder("path/to/proto", outputTable,3,1,0.1);
coder.parseAndHandle(data); //data 为只有一列 string 列的表，存储 proto 原始数据，输出到了 outputTable 表中