需要掌握：flink1.12.x Table API 和 SQL：创建工程，读取csv文件，创建表，实现查询（静态表）
介绍
本文实现查询一个表，并将数据写入到另一个表中，保存到文件系统中
非聚合查询
如果是 聚合查询，来一个数据，计算一个数据，需要 Retract 和 Add 操作，，而文件系统 只支持 append 操作，所以会报错
java
package test;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
public class TestTable读取文件_输出到文件sql {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
//构建Table环境
EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, bsSettings);
/*
注意不要出现关键字，如：order等
需要依赖flink-csv
*/
TableResult table=stEnv.executeSql("CREATE TABLE t_order (" +
" name STRING," +
" price double," +
" username STRING" +
") WITH ( " +
" 'connector' = 'filesystem'," +
" 'path' = 'file:///D:/bigdata/flink_table_sql/data'," +
" 'format' = 'csv'," +
" 'csv.field-delimiter' = ','" +
")");
/*
创建结果表
*/
TableResult sinkTable=stEnv.executeSql("CREATE TABLE sink_order (" +
" username STRING," +
" price double" +
") WITH ( " +
" 'connector' = 'filesystem'," +
" 'path' = 'file:///D:/bigdata/flink_table_sql/sink'," +
" 'csv.disable-quote-character' ='true'," +
" 'format' = 'csv'," +
" 'csv.field-delimiter'=','" +
")");
//执行查询
Table res=stEnv.sqlQuery("select username,price from t_order where price>60");
System.out.println("Schema信息:");
//打印Schema
res.printSchema();
/*
不写报错
No operators defined in streaming topology. Cannot execute.
*/
DataStream<Tuple2<Boolean, Row>> dsRes = stEnv.toRetractStream(res, Row.class);
dsRes.print();
stEnv.executeSql("insert into sink_order select * from "+res);
env.execute();
}
}