1.2 基于DataStream api的sql

示例代码

package org.apache.flink.table.examples.java;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.Table;import org.apache.flink.table.api.java.StreamTableEnvironment;import java.util.Arrays;/** * Simple example for demonstrating the use of SQL on a Stream Table in Java.

*

* <p>This example shows how to:

* - Convert DataStreams to Tables

* - Register a Table under a name

* - Run a StreamSQL query on the registered Table

*

*/publicclass StreamSQLExample {

// *************************************************************************

// PROGRAM

// *************************************************************************publicstaticvoidmain(String[] args)throws Exception {

// set up execution environmentStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

DataStream orderA = env.fromCollection(Arrays.asList(

newOrder(1L, "beer", 3),

newOrder(1L, "diaper", 4),

newOrder(3L, "rubber", 2)));

DataStream orderB = env.fromCollection(Arrays.asList(

newOrder(2L, "pen", 3),

newOrder(2L, "rubber", 3),

newOrder(4L, "beer", 1)));

// convert DataStream to TableTable tableA = tEnv.fromDataStream(orderA, "user, product, amount");

// register DataStream as TabletEnv.registerDataStream("OrderB", orderB, "user, product, amount");

// union the two tablesTable result = tEnv.sqlQuery("SELECT * FROM " + tableA + " WHERE amount > 2 UNION ALL " + "SELECT * FROM OrderB WHERE amount < 2"); tEnv.toAppendStream(result, Order.class).print();env.execute(); }// *************************************************************************

// USER DATA TYPES

// *************************************************************************/** * Simple POJO.

*/publicstaticclass Order {

public Long user;

public String product;

publicint amount;

public Order() {

}

publicOrder(Long user, String product,int amount) {

this.user = user;

this.product = product;

this.amount = amount;

}

@Override

public String toString() {

return"Order{" + "user=" + user + ", product='" + product + '\'' + ", amount=" + amount + '}';

}

}

}

其中，StreamTableEnvironment

/** * The {@link TableEnvironment} for a Java {@link StreamExecutionEnvironment} that works with

* {@link DataStream}s.

*

* <p>A TableEnvironment can be used to:

* <ul>

* <li>convert a {@link DataStream} to a {@link Table}</li>

* <li>register a {@link DataStream} in the {@link TableEnvironment}'s catalog</li>

* <li>register a {@link Table} in the {@link TableEnvironment}'s catalog</li>

* <li>scan a registered table to obtain a {@link Table}</li>

* <li>specify a SQL query on registered tables to obtain a {@link Table}</li>

* <li>convert a {@link Table} into a {@link DataStream}</li>

* <li>explain the AST and execution plan of a {@link Table}</li>

* </ul>

*/

 StreamTableSource

/** Defines an external stream table and provides read access to its data.

*

* @param <T> Type of the {@link DataStream} created by this {@link TableSource}.

*/

StreamTableSink

/** * Defines an external stream table and provides write access to its data.

*

* @param <T> Type of the {@link DataStream} created by this {@link TableSink}.

*/

2. table api

示例

package org.apache.flink.table.examples.java;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;/** * Simple example for demonstrating the use of the Table API for a Word Count in Java.

*

* <p>This example shows how to:

* - Convert DataSets to Tables

* - Apply group, aggregate, select, and filter operations

*/publicclass WordCountTable {

// *************************************************************************

// PROGRAM

// *************************************************************************publicstaticvoidmain(String[] args)throws Exception {

ExecutionEnvironment env = ExecutionEnvironment.createCollectionsEnvironment();

BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

DataSet input = env.fromElements(

newWC("Hello", 1),

newWC("Ciao", 1),

newWC("Hello", 1));

Table table = tEnv.fromDataSet(input);

Table filtered = table

.groupBy("word")

.select("word, frequency.sum as frequency")

.filter("frequency = 2");DataSet<WC> result = tEnv.toDataSet(filtered, WC.class);result.print(); }// *************************************************************************

// USER DATA TYPES

// *************************************************************************/** * Simple POJO containing a word and its respective count.

*/publicstaticclass WC {

public String word;

publiclong frequency;

// public constructor to make it a Flink POJOpublic WC() {}

publicWC(String word,long frequency) {

this.word = word;

this.frequency = frequency;

}

@Override

public String toString() {

return"WC " + word + " " + frequency;

}

}

}

 3.数据转换

　　3.1 DataSet与Table相互转换

　　　　DataSet-->Table

　　　　　　注册方式：

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

转换方式：

Table table = tEnv.fromDataSet(input);

　　　　Table-->DataSet

　　　　　　　　DataSet<WC> result = tEnv.toDataSet(filtered, WC.class);

　　3.2 DataStream与Table相互转换

　　　　DataStream-->Table

　　　　　　注册方式：

　　tEnv.registerDataStream("OrderB", orderB, "user, product, amount");

转换方式：

Table tableA = tEnv.fromDataStream(orderA, "user, product, amount");

　　　　Table-->DataStream

　　　　　　　　DataSet<WC> result = tEnv.toDataSet(filtered, WC.class);