提供了编程的基础技术教程

zl程序教程

您现在的位置是：首页 >  其他

当前栏目

Flink中DataStream和Table互相转换

2023-02-25 18:17:32 时间

前言

Flink 为处理一列转多列的场景提供了两种返回类型 Tuple 和 Row

  • Tuple 只支持1~25个字段，且不能为null，不支持拓展
  • Row 支持null同时也无限制字段数，但如果需要使用Row，必须重载实现getResultType方法

DataStream=>Table

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;

public class TCS002 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        List<Row> list = new ArrayList<Row>();
        list.add(Row.of("张三",18,"男"));
        list.add(Row.of("xiaohong",16,"女"));

        RowTypeInfo rowTypeInfo = getRowTypeInfo(list.get(0));
        DataStream<Row> ds = env.fromCollection(list,rowTypeInfo);
        tabEnv.registerDataStream("table01",ds);
        tabEnv.from("table01").execute().print();
    }

    private static RowTypeInfo getRowTypeInfo(Row row) {
        TypeInformation[] types = new TypeInformation[row.getArity()];
        String[] fieldNames = new String[row.getArity()];
        for (int i = 0; i < row.getArity(); i++) {
            Object field = row.getField(i);
            if(field instanceof Integer){
                types[i] = BasicTypeInfo.INT_TYPE_INFO;
            }else{
                types[i] = BasicTypeInfo.STRING_TYPE_INFO;
            }
            fieldNames[i] = "f"+i;
        }
        return new RowTypeInfo(types, fieldNames);
    }
}

Table=>DataStream

DataStream<Row> ds02 = tabEnv.toAppendStream(tb01, rowTypeInfo);

相关文章