CsvToTriple

功能介绍

将数据格式从 Csv 转成 Triple

参数说明

名称 中文名称 描述 类型 是否必须？ 默认值
handleInvalid 解析异常处理策略 解析异常处理策略 String “ERROR”
tripleColumnValueSchemaStr 三元组结构中列信息和数据信息的Schema 三元组结构中列信息和数据信息的Schema String ✓
reservedCols 算法保留列名 算法保留列 String[] []
csvCol CSV列名 CSV列的列名 String ✓
schemaStr Schema Schema。格式为”colname coltype[, colname2, coltype2[, …]]”，例如”f0 string, f1 bigint, f2 double” String ✓
csvFieldDelimiter 字段分隔符 字段分隔符 String “,”
quoteChar 引号字符 引号字符 Character “\””

脚本示例

脚本代码

  1. import numpy as np
  2. import pandas as pd
  5. data = np.array([['1', '{"f1":"1.0","f2":"2.0"}', '$3$1:1.0 2:2.0', '1:1.0,2:2.0', '1.0,2.0', 1.0, 2.0],
  6. ['2', '{"f2":"4.0","f4":"8.0"}', '$3$1:4.0 2:8.0', '1:4.0,2:8.0', '4.0,8.0', 4.0, 8.0]])
  8. df = pd.DataFrame({"row":data[:,0], "json":data[:,1], "vec":data[:,2], "kv":data[:,3], "csv":data[:,4], "f0":data[:,5], "f1":data[:,6]})
  9. data = dataframeToOperator(df, schemaStr="row string, json string, vec string, kv string, csv string, f0 double, f1 double",op_type="stream")
  12. op = CsvToTripleStreamOp()\
  13.     .setCsvCol("csv").setSchemaStr("f0 double, f1 double")\
  14.     .setReservedCols(["row"]).setTripleColValSchemaStr("col string, val double")\
  15.     .linkFrom(data)
  16. op.print()
  17. StreamOperator.execute()

脚本运行结果

  1. |row|col|val|
  2. |-|-|---|
  3. |1|1|1.0|
  4. |1|2|2.0|
  5. |2|1|4.0|
  6. |2|2|8.0|