FLink Table API JAVA_BATCH_DEMO

Posted 2020-11-23 maoxiangyi

<dependencies>
    <!-- Either... -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_2.11</artifactId>
        <version>1.9.0</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
        <version>1.9.0</version>
        <!--<scope>provided</scope>-->
    </dependency>

    <!-- Either... (for the old planner that was available before Flink 1.9) -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner_2.11</artifactId>
        <version>1.9.0</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_2.11</artifactId>
        <version>1.9.0</version>
        <!--<scope>provided</scope>-->
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.11</artifactId>
        <version>1.9.0</version>
        <!--<scope>provided</scope>-->
    </dependency>

</dependencies>

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;

/**
 *
 * FLink Java Batch Table API DEMO
 *
 * @author: create by maoxiangyi
 * @version: v1.0
 */
public class WordCountSql {

   public static void main(String[] args) throws Exception {
      ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
      BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);

      DataSet<WC> input = fbEnv.fromElements(
            new WC("Hello", 1),
            new WC("Ciao", 1),
            new WC("Hello", 1)
      );


      // register the DataSet as table "WordCount"
      fbTableEnv.registerDataSet("WordCount", input, "word, frequency");

      // run a SQL query on the Table and retrieve the result as a new Table
      Table table = fbTableEnv.sqlQuery(
            "SELECT word, SUM(frequency) as frequency FROM WordCount GROUP BY word");

      DataSet<com.xesv5.mxy.WC> result = fbTableEnv.toDataSet(table, com.xesv5.mxy.WC.class);

      result.print();
   }
}