Flink DateStreamまたはDateSetとTableの相互変換(Java)

@羲凡——もっとよく生きるためだけに
Flink DateStream/DateSetとTableの相互変換
一.前提準備
pom.xmlファイルに依存を追加するには

<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-table_2.11artifactId>
    <version>${flink.version}version>
dependency>

二.DateSetとTableの相互変換

package test;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkSqlTest01 {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
        DataSource<Tuple2> dataSource = env.fromElements(
                new Tuple2("  ", "    "),
                new Tuple2("  ", "    "),
                new Tuple2("  ", "    "));
        tableEnv.registerDataSet("test", dataSource, "name,home");
        Table table = tableEnv.sqlQuery("select name from test");
        table.printSchema();
        DataSet<Row> rowDataSet = tableEnv.toDataSet(table, Row.class);
        rowDataSet.print();
    }
}

三.DateStreamとTableの相互変換

package test;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkSqlTest02 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(streamEnv);
        DataStreamSource<Tuple2> dataStreamSource = streamEnv.fromElements(
                new Tuple2("  ", "    "),
                new Tuple2("  ", "    "),
                new Tuple2("  ", "    "));
        tableEnv.registerDataStream("test", dataStreamSource, "name,home");
        Table table = tableEnv.sqlQuery("select * from test");
        table.printSchema();
        /*DataStream rowDataStream = tableEnv.toAppendStream(table, Row.class);
        rowDataStream.print();*/
        DataStream<Tuple2<Boolean, Row>> dataStream = tableEnv.toRetractStream(table, Row.class);
        dataStream.print();
        streamEnv.execute();
    }
}

====================================================================
@羲凡——もっとよく生きるためだけに
ブログに何か問題があれば、メッセージ交換を歓迎します.