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

功能介绍

该组件将 MTable 流数据展开成 Table。

参数说明

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

| schemaStr | Schema | Schema。格式为”colname coltype[, colname2, coltype2[, …]]”，例如”f0 string, f1 bigint, f2 double” | String | ✓ | | |

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

| handleInvalidMethod | 处理无效值的方法 | 处理无效值的方法，可取 error, skip | String | | “ERROR”, “SKIP” | “ERROR” |

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

代码示例

Python 代码

  1. import numpy as np
  2. import pandas as pd
  3. df_data = pd.DataFrame([
  4.       ["a1", "{\"data\":{\"f0\":[\"11L\",\"12L\"],\"f1\":[2.0,2.0]},\"schema\":\"f0 VARCHAR,f1 DOUBLE\"}"],
  5.       ["a1", "{\"data\":{\"f0\":[\"13L\",\"14L\"],\"f1\":[2.0,2.0]},\"schema\":\"f0 VARCHAR,f1 DOUBLE\"}"]
  6. ])
  8. input = StreamOperator.fromDataframe(df_data, schemaStr='id string, mt string')
  10. flatten = FlattenMTableStreamOp()\
  11.     .setReservedCols(["id"])\
  12.     .setSelectedCol("mt")\
  13.     .setSchemaStr('f0 string, f1 int')
  15. input.link(flatten).print()
  16. StreamOperator.execute()

Java 代码

  1. package com.alibaba.alink.operator.batch.dataproc;
  3. import org.apache.flink.types.Row;
  5. import com.alibaba.alink.operator.stream.StreamOperator;
  6. import com.alibaba.alink.operator.stream.dataproc.FlattenMTableStreamOp;
  7. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  8. import com.alibaba.alink.testutil.AlinkTestBase;
  9. import org.junit.Test;
  11. import java.util.ArrayList;
  12. import java.util.List;
  14. /**
  15.  * Test cases for gbdt.
  16.  */
  18. public class FlattenMTableTest extends AlinkTestBase {
  20.     @Test
  21.     public void test3() throws Exception {
  22.         List <Row> rows = new ArrayList <>();
  23.         rows.add(Row.of("a1", "{\"data\":{\"f0\":[\"11L\",\"12L\"],\"f1\":[2.0,2.0]},\"schema\":\"f0 VARCHAR,f1 "
  24.             + "DOUBLE\"}"));
  25.             rows.add(Row.of("a3", "{\"data\":{\"f0\":[\"13L\",\"14L\"],\"f1\":[2.0,2.0]},\"schema\":\"f0 VARCHAR,f1 "
  26.                 + "DOUBLE\"}"));
  27.         StreamOperator <?> input = new MemSourceStreamOp(rows, "id string, mt string");
  29.         FlattenMTableStreamOp flatten = new FlattenMTableStreamOp()
  30.             .setReservedCols("id")
  31.             .setSelectedCol("mt")
  32.             .setSchemaStr("f0 string, f1 int");
  33.         flatten.linkFrom(input).print();
  34.         StreamOperator.execute();
  35.     }
  36. }

运行结果

| id | f0 | f1 | | —- | —- | —- |

| a3 | 13L | 2 |

| a1 | 11L | 2 |

| a1 | 12L | 2 |

| a3 | 14L | 2 |