0.4 TableSink

数据处理完成后需要将结果写入外部存储中，在Table API中有对应的Sink模块，此模块为TableSink

/** * A {@link TableSink} specifies how to emit a table to an external

* system or location.

*

* <p>The interface is generic such that it can support different storage locations and formats.

*

* @param <T> The return type of the {@link TableSink}.

*/

0.5 Table Connector

在Flink1.6版本之后，为了能够让Table API通过配置化的方式连接外部系统，且同时可以在sql client中使用，flink 提出了Table Connector的概念，主要目的时将Table Source和Table Sink的定义和使用分离。

通过Table Connector将不同内建的Table Source和TableSink封装，形成可以配置化的组件，在Table Api和Sql client能够同时使用。

/** * Creates a table source and/or table sink from a descriptor.

*

* <p>Descriptors allow for declaring the communication to external systems in an

* implementation-agnostic way. The classpath is scanned for suitable table factories that match

* the desired configuration.

*

* <p>The following example shows how to read from a connector using a JSON format and

* register a table source as "MyTable":

*

* <pre>

* {@code *

* tableEnv

* .connect(

* new ExternalSystemXYZ()

* .version("0.11"))

* .withFormat(

* new Json()

* .jsonSchema("{...}")

* .failOnMissingField(false))

* .withSchema(

* new Schema()

* .field("user-name", "VARCHAR").from("u_name")

* .field("count", "DECIMAL")

* .registerSource("MyTable");

* }

*</pre>

*

* @param connectorDescriptor connector descriptor describing the external system

*/ TableDescriptor connect(ConnectorDescriptor connectorDescriptor);

 本篇主要聚焦于sql和Table Api。

 1.sql

1.1 基于DataSet api的sql

示例：

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 that shows how the Batch SQL API is used in Java.

*

* <p>This example shows how to:

* - Convert DataSets to Tables

* - Register a Table under a name

* - Run a SQL query on the registered Table

*/publicclass WordCountSQL {

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

// PROGRAM

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

// set up execution environmentExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

DataSet input = env.fromElements(

newWC("Hello", 1),

newWC("Ciao", 1),

newWC("Hello", 1));

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

// run a SQL query on the Table and retrieve the result as a new TableTable table = tEnv.sqlQuery(

"SELECT word, SUM(frequency) as frequency FROM WordCount GROUP BY word");

DataSet result = tEnv.toDataSet(table, 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;

}

}

}

其中，BatchTableEnvironment

/** * The {@link TableEnvironment} for a Java batch {@link ExecutionEnvironment} that works

* with {@link DataSet}s.

*

* <p>A TableEnvironment can be used to:

* <ul>

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

* <li>register a {@link DataSet} 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 DataSet}</li>

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

* </ul>

*/

BatchTableSource

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

*

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

*/

BatchTableSink

/** Defines an external {@link TableSink} to emit a batch {@link Table}.

*

* @param <T> Type of {@link DataSet} that this {@link TableSink} expects and supports.

*/