flink1.12.x Table API 和 SQL：Row、RowTypeInfo、TypeInformation、BasicTypeInfo

作者：马育民 • • 阅读：10163

作用

  • 在source中，封装一行数据，指定 一行数据的列名和列类型

  • 将 查询结果（Table） 转换成 DataStream 时，可以将一行查询结果封装成 Row 对象中

常用方法

获取指定列的值

getField(int pos)

获取类型

row.getKind()

有以下分类：

INSERT("+I", (byte)0),
DELETE("-D", (byte)3),

UPDATE_BEFORE("-U", (byte)1), 聚合时，来一条数据，重新计算，表示删除上一条记录
UPDATE_AFTER("+U", (byte)2),  聚合时，来一条数据，重新计算，表示增加一条记录

获取列数

getArity()

source中，封装一行数据

在 Table API 和 SQL 中，Row 可以封装一行数据

应用场景

没用 connector 连接器 和 建表语句

关键代码

// 解析数据
SingleOutputStreamOperator<Row> soso = dss.flatMap(new FlatMapFunction<String, Row>() {
    @Override
    public void flatMap(String value, Collector<Row> out) throws Exception {
        String[] arr = value.split(",");
        Double price = Double.parseDouble(arr[1]);
        out.collect(Row.of(arr[0],price,arr[2]));
    }
});

// 指定各个字段类型
TypeInformation[] types = new TypeInformation[]{
        BasicTypeInfo.STRING_TYPE_INFO,
        BasicTypeInfo.DOUBLE_TYPE_INFO,
        BasicTypeInfo.STRING_TYPE_INFO
};

// 指定各个字段名称
String[] fieldNames = new String[]{"name","price","username"};

// 指定 row中各个字段名称和类型
RowTypeInfo rowTypeInfo = new RowTypeInfo(types, fieldNames);
SingleOutputStreamOperator<Row> typeInfoSOSO = soso.returns(rowTypeInfo);

例子

package p2socket;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
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.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class C4Row {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);

        DataStreamSource<String> dss = env.fromElements(
                "java,50,lilei",
                "hadoop,90,lilei",
                "flink,70,lilei"
                );

        SingleOutputStreamOperator<Row> soso = dss.flatMap(new FlatMapFunction<String, Row>() {
            @Override
            public void flatMap(String value, Collector<Row> out) throws Exception {
                String[] arr = value.split(",");
                Double price = Double.parseDouble(arr[1]);
                out.collect(Row.of(arr[0],price,arr[2]));
            }
        });

        // 指定各个字段类型
        TypeInformation[] types = new TypeInformation[]{
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.DOUBLE_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO
        };
        // 指定各个字段名称
        String[] fieldNames = new String[]{"name","price","username"};
        // 指定 row中各个字段名称和类型
        RowTypeInfo rowTypeInfo = new RowTypeInfo(types, fieldNames);

        SingleOutputStreamOperator<Row> typeInfoSOSO = soso.returns(rowTypeInfo);

        stEnv.createTemporaryView("v_order",typeInfoSOSO);
        Table res=stEnv.sqlQuery("select name,price from v_order where price>60");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();
        /*
        将查询结果，封装到 Row 类对象中
         */
        DataStream<Row> dsRes = stEnv.toAppendStream(res, Row.class);
        dsRes.print();
        env.execute();
    }
}

查询结果，封装一行数据

非聚合查询

package test;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Rowlijie2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);
        /*
        注意不要出现关键字，如：order等
        需要依赖flink-csv
         */
        TableResult table=stEnv.executeSql("CREATE TABLE t_order (" +
                "  name STRING," +
                "  price double," +
                "  username STRING" +
                ") WITH ( " +
                "  'connector' = 'filesystem'," +
                "  'path' = './data/flink.csv'," +
                "  'format' = 'csv'," +
                "  'csv.field-delimiter' = ','" +
                ")");
        //执行查询
        Table res=stEnv.sqlQuery("select username,price from t_order where price>60");
        System.out.println("Schema信息:");
        // 打印Schema
        res.printSchema();

        DataStream<Row> dsRes = stEnv.toAppendStream(res, Row.class);
        dsRes.map(new MapFunction<Row, Object>() {
            @Override
            public Object map(Row value) throws Exception {
                System.out.println(value.getKind()+"-"+value.getField(0)+"-"+value.getField(1));
                return value;
            }
        });

        env.execute("myjob");
    }
}

执行结果如下：

INSERT-李雷-80.0
INSERT-韩梅梅-90.0
INSERT-韩梅梅-70.0
INSERT-李雷-100.0
INSERT-lucy-100.0
INSERT-lucy-120.0

聚合查询

package test;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Rowlijie3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        //构建Table环境
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);
        /*
        注意不要出现关键字，如：order等
        需要依赖flink-csv
         */
        TableResult table=stEnv.executeSql("CREATE TABLE t_order (" +
                "  name STRING," +
                "  price double," +
                "  username STRING" +
                ") WITH ( " +
                "  'connector' = 'filesystem'," +
                "  'path' = './data/flink.csv'," +
                "  'format' = 'csv'," +
                "  'csv.field-delimiter' = ','" +
                ")");
        //执行查询
        Table res=stEnv.sqlQuery("select username,avg(price) from t_order where price>60 group by username");
        System.out.println("Schema信息:");
        //打印Schema
        res.printSchema();
        /*
        需要将结果转换成DataStream，才能打印结果，需要指定 封装的泛型
        将计算后的数据，在ds原数据基础上更新
        true：表示更新
        false：表示删除
         */
        DataStream<Tuple2<Boolean, Row>> dsRes = stEnv.toRetractStream(res, Row.class);

        dsRes.map(new MapFunction<Tuple2<Boolean, Row>, Object>() {
            @Override
            public Object map(Tuple2<Boolean, Row> data) throws Exception {
                Row row = data.f1;
                System.out.println(data.f0+"-"+row.getKind()+"-"+row.getField(0)+"-"+row.getField(1));
                return row;
            }
        });

        env.execute("myjob");
    }
}

执行结果如下：

true-INSERT-李雷-80.0
true-INSERT-韩梅梅-90.0
false-UPDATE_BEFORE-韩梅梅-90.0 // 删除该数据
true-UPDATE_AFTER-韩梅梅-80.0 // 插入该数据
false-UPDATE_BEFORE-李雷-80.0
true-UPDATE_AFTER-李雷-90.0
true-INSERT-lucy-100.0
false-UPDATE_BEFORE-lucy-100.0
true-UPDATE_AFTER-lucy-110.0

原文出处：/show_1IX4E5u42q5C.html

吉ICP备15003043号