目录

API

获取环境

创建表

查询表

Table API

SQL

​​​​​​​写出表

​​​​​​​与DataSet/DataStream集成

​​​​​​​TableAPI

​​​​​​​SQLAPI

API

获取环境

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/common.html#create-a-tableenvironment

      // **********************
      // FLINK STREAMING QUERY
      // **********************
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.table.api.EnvironmentSettings;
      import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
      EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
      StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
      StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
      // or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);
      // ******************
      // FLINK BATCH QUERY
      // ******************
      import org.apache.flink.api.java.ExecutionEnvironment;
      import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
      ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
      BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);
      // **********************
      // BLINK STREAMING QUERY
      // **********************
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      import org.apache.flink.table.api.EnvironmentSettings;
      import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
      StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
      EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
      StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
      // or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
      // ******************
      // BLINK BATCH QUERY
      // ******************
      import org.apache.flink.table.api.EnvironmentSettings;
      import org.apache.flink.table.api.TableEnvironment;
      EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
      TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);
  
 

​​​​​​​创建表

      // get a TableEnvironment
      TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // table is the result of a simple projection query
      Table projTable = tableEnv.from("X").select(...);
      // register the Table projTable as table "projectedTable"
      tableEnv.createTemporaryView("projectedTable", projTable);
  
 

      tableEnvironment
        .connect(...)
        .withFormat(...)
        .withSchema(...)
        .inAppendMode()
        .createTemporaryTable("MyTable")
  
 

​​​​​​​查询表

Table API

      // get a TableEnvironment
      TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // register Orders table
      // scan registered Orders table
      Table orders = tableEnv.from("Orders");// compute revenue for all customers from France
      Table revenue = orders
        .filter($("cCountry")
      .isEqual("FRANCE"))
        .groupBy($("cID"), $("cName")
        .select($("cID"), $("cName"), $("revenue")
      .sum()
      .as("revSum"));
      // emit or convert Table
      // execute query
  
 

SQL

      // get a TableEnvironment
      TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // register Orders table
      // compute revenue for all customers from France
      Table revenue = tableEnv.sqlQuery(
          "SELECT cID, cName, SUM(revenue) AS revSum " +
          "FROM Orders " +
          "WHERE cCountry = 'FRANCE' " +
          "GROUP BY cID, cName"
        );
      // emit or convert Table
      // execute query
  
 

      // get a TableEnvironment
      TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // register "Orders" table
      // register "RevenueFrance" output table
      // compute revenue for all customers from France and emit to "RevenueFrance"
      tableEnv.executeSql(
          "INSERT INTO RevenueFrance " +
          "SELECT cID, cName, SUM(revenue) AS revSum " +
          "FROM Orders " +
          "WHERE cCountry = 'FRANCE' " +
          "GROUP BY cID, cName"
        );
  
 

​​​​​​​写出表

      // get a TableEnvironment
      TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // create an output Table
      final Schema schema = new Schema()
          .field("a", DataTypes.INT())
          .field("b", DataTypes.STRING())
          .field("c", DataTypes.BIGINT());
      tableEnv.connect(new FileSystem().path("/path/to/file"))
          .withFormat(new Csv().fieldDelimiter('|').deriveSchema())
          .withSchema(schema)
          .createTemporaryTable("CsvSinkTable");
      // compute a result Table using Table API operators and/or SQL queries
      Table result = ...
      // emit the result Table to the registered TableSink
      result.executeInsert("CsvSinkTable");
  
 

​​​​​​​与DataSet/DataStream集成

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/common.html#integration-with-datastream-and-dataset-api

• Create a View from a DataStream or DataSet

      // get StreamTableEnvironment
      // registration of a DataSet in a BatchTableEnvironment is equivalent
      StreamTableEnvironment tableEnv = ...;
      // see "Create a TableEnvironment" section
      DataStream<Tuple2<Long, String>> stream = ...
      // register the DataStream as View "myTable" with fields "f0", "f1"
      tableEnv.createTemporaryView("myTable", stream);
      // register the DataStream as View "myTable2" with fields "myLong", "myString"
      tableEnv.createTemporaryView("myTable2", stream, $("myLong"), $("myString"));
  
 

• Convert a DataStream or DataSet into a Table

  
          // get StreamTableEnvironment// registration of a DataSet in a BatchTableEnvironment is equivalent
          StreamTableEnvironment tableEnv = ...;
          // see "Create a TableEnvironment" section
          DataStream<Tuple2<Long, String>> stream = ...
          // Convert the DataStream into a Table with default fields "f0", "f1"
          Table table1 = tableEnv.fromDataStream(stream);
          // Convert the DataStream into a Table with fields "myLong", "myString"
          Table table2 = tableEnv.fromDataStream(stream, $("myLong"), $("myString"));
      
     
  

• Convert a Table into a DataStream or DataSet

Convert a Table into a DataStream

Append Mode: This mode can only be used if the dynamic Table is only modified by INSERT changes, i.e, it is append-only and previously emitted results are never updated.

追加模式：只有当动态表仅通过插入更改进行修改时，才能使用此模式，即，它是仅追加模式，并且以前发出的结果从不更新。

Retract Mode: This mode can always be used. It encodes INSERT and DELETE changes with a boolean flag.

撤回模式：此模式始终可用。它使用布尔标志对插入和删除更改进行编码。

      // get StreamTableEnvironment.
      StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section
      // Table with two fields (String name, Integer age)
      Table table = ...
      // convert the Table into an append DataStream of Row by specifying the class
      DataStream<Row> dsRow = tableEnv.toAppendStream(table, Row.class);
      // convert the Table into an append DataStream of Tuple2<String, Integer>
       //   via a TypeInformation
      TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
        Types.STRING(),
        Types.INT());
      DataStream<Tuple2<String, Integer>> dsTuple =
        tableEnv.toAppendStream(table, tupleType);
      // convert the Table into a retract DataStream of Row.
      //   A retract stream of type X is a DataStream<Tuple2<Boolean, X>>.
      //   The boolean field indicates the type of the change.
      //   True is INSERT, false is DELETE.
      DataStream<Tuple2<Boolean, Row>> retractStream =
        tableEnv.toRetractStream(table, Row.class);
  
 

Convert a Table into a DataSet

      // get BatchTableEnvironment
      BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
      // Table with two fields (String name, Integer age)
      Table table = ...
      // convert the Table into a DataSet of Row by specifying a class
      DataSet<Row> dsRow = tableEnv.toDataSet(table, Row.class);
      // convert the Table into a DataSet of Tuple2<String, Integer> via a TypeInformationTupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
        Types.STRING(),
        Types.INT());
      DataSet<Tuple2<String, Integer>> dsTuple =
        tableEnv.toDataSet(table, tupleType);
  
 

​​​​​​​TableAPI

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/tableApi.html

​​​​​​​SQLAPI

https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/sql/

文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。

原文链接：lansonli.blog.csdn.net/article/details/116327559