Flink SQL 批處理

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner_2.11</artifactId>
        <version>1.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-java-bridge_2.11</artifactId>
        <version>1.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
        <version>1.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.11</artifactId>
        <version>1.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-common</artifactId>
        <version>1.9.0</version>
    </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;

public class FLinkSqlBatch {
   public static void main(String[] args) throws Exception {
      //1) 獲取執行環境
      ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
      BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
      /**
       * 1.csv
       * channel,subject,refer,reg,ord,pv,uv
       * 朋友圈,三科,H5,100,100,100,100
       * 朋友圈,數學,H5,100,100,100,100
       * 朋友圈,理科,H5,100,100,100,100
       * 朋友圈,編程,H5,100,100,100,100
       * 朋友圈,英語,H5,100,100,100,100
       * 朋友圈,通用,H5,100,100,100,100
       */
      // 2）讀取數據
      DataSet<AdPojo> csvInput = env
            .readCsvFile("D:\\code\\learn\\flink-sql\\src\\main\\resources\\1.csv")
            .ignoreFirstLine() //忽略第一行
            .pojoType(AdPojo.class, "channel", "subject", "refer", "reg", "ord", "pv", "uv");

      //3）將DataSet轉換為Table，並注冊為table1
      Table topScore = tableEnv.fromDataSet(csvInput);
      tableEnv.registerTable("table1", topScore);

      //4）自定義sql語句
      Table groupedByCountry = tableEnv.sqlQuery("select channel,subject,refer,reg,ord,pv,uv from table1");
      //5）轉換回dataset
      DataSet<AdPojo> result = tableEnv.toDataSet(groupedByCountry, AdPojo.class);
      //6）打印
      result.print();
   }
}

AdPojo{channel='朋友圈', subject='英語', refer='H5', reg='100', ord='100', pv='100', uv='100'}
AdPojo{channel='朋友圈', subject='通用', refer='H5', reg='100', ord='100', pv='100', uv='100'}
AdPojo{channel='朋友圈', subject='三科', refer='H5', reg='100', ord='100', pv='100', uv='100'}
AdPojo{channel='朋友圈', subject='理科', refer='H5', reg='100', ord='100', pv='100', uv='100'}
AdPojo{channel='朋友圈', subject='數學', refer='H5', reg='100', ord='100', pv='100', uv='100'}
AdPojo{channel='朋友圈', subject='編程', refer='H5', reg='100', ord='100', pv='100', uv='100'}