Flink中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);
    }
}

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