Flink系统之Table API 和 SQL

golang中使用Shutdown特性对http服务进行优雅退出使用总结

　　Flink供应了像表一样处置惩罚的API和像实行SQL语句一样把效果集举行实行。如许很轻易的让人人举行数据处置惩罚了。比方实行一些查询，在无界数据和批处置惩罚的使命上，然后将这些按肯定的花样举行输出，很轻易的让人人像实行SQL一样简朴。

　　本日重要写的东西分为以下几个方面，然后遵照着下边几个方面举行睁开：

　　1. Flink的差别API的层级梗概。

　　2. FlinkSQL的编程的步骤。

　　3. Flink编程的例子。

一、  Flink有着差别级别的API，差别级别的API轻易差别用户举行处置惩罚。普通用户运用Datastream以及Dataset举行程序编写，我们能够在其更高的基础上运用Table API以及SQL，这也是Flink的壮大的地方，能够像运用处置惩罚表一样处置惩罚数据。假如想研讨的更高能够看更底层的东西。

二、 Flink的Table API 和 SQL编程步骤以下：

　　1） 建立一个TableEnvironment表环境用于后续运用。TableEnvironment是 SQL 和 Table API的中心观点，它用于设置实行所须要的数据属性，和ExecutionEnvironment相似，它重要担任：

　　　　a) 注册表数据源，从内部或许外部泉源。

　　　　b) 实行响应的SQL语句。

　　　　c) 注册自定义集数。

　　　　d 将效果集举行扫描和写入到目的数据源。

　　　　e) 雷同的environment能够实行响应的join unin操纵。

　　2）接下来，我们看一下怎样注册数据源，注重差别的Flink版本有差别的完成，然则中心的内容是稳定的：

　　　　a) 能够直接从数据集里举行注册。比方 tableEnvironment.registerDataSet()。

　　　　b) 在一个已存在的Table中直接实行scan或许select，那末会生成一个新的Table，也就是数据能够从已有的Table中再次猎取，Table t = tableEnv.scan("x").select("a, b,c")。

　　　　c) 能够是TableSource, 也就是从差别的文件、数据库、音讯体系举行读取。 比方csv文件，TableSource csvSource = new CsvTableSource("path/to/file")。

　　3）读取完数据后举行处置惩罚，处置惩罚完以后要存储起来，那末须要Sink（存储）到文件或许数据库、音讯体系等。

　　　　a) 比方Sink到CSV文件。 TableSink csvSink = new TableCSVSink("path/to/sink", ..)。

　　　　b) Sink为指定字段句和范例到CSV文件中。

　　　　　　指定表字段： String[] fieldNames = {"fild1", "filed2", "field3"}; 

　　　　　　指定字段范例： TypeInformation[] fieldTypes = {Types.INT, Types.STRING, Types.LONG}; 

　　　　　　指定表名和csv文件：tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, csvSink);

三、接下来，看一下实在的例子。


　　  1）从给定的单词和单词的个数中统计一下，每一个单词涌现的数据，运用SQL语句举行完成查询统计。完全的样例以下（注重，差别的FLink版本完成上有轻微的差别）：

package myflink.sql;

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 WordCountSQL {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = BatchTableEnvironment.getTableEnvironment(env);

        DataSet<WC> input = env.fromElements(
                WC.of("hello", 1),
                WC.of("hqs", 1),
                WC.of("world", 1),
                WC.of("hello", 1)
        );
        //注册数据集
        tEnv.registerDataSet("WordCount", input, "word, frequency");

        //实行SQL，并效果集做为一个新表
        Table table = tEnv.sqlQuery("SELECT word, SUM(frequency) as frequency FROM WordCount GROUP BY word");

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

        result.print();

    }

    public static class WC {
        public String word; //hello
        public long frequency;

        //建立组织要领，让flink举行实例化
        public WC() {}

        public static WC of(String word, long frequency) {
            WC wc = new WC();
            wc.word = word;
            wc.frequency = frequency;
            return wc;
        }

        @Override
        public String toString() {
            return "WC " + word + " " + frequency;
        }
    }

}

　　输出的效果为，和我们想的效果是一样的。

WC world 1
WC hello 2
WC hqs 1

2）接下来的例子会庞杂一些，从一个txt文件中读取数据，txt文件中包括id, 人字, 书名，价钱信息。然后将数据注册成一个表，然后将这个表的效果举行统计，按人名统计出来这个人买书所消费的钱，将效果sink到一个文件中。上代码。

package myflink.sql;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sinks.TableSink;

public class SQLFromFile {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        BatchTableEnvironment tableEnv = BatchTableEnvironment.getTableEnvironment(env);

        env.setParallelism(1);
        //读取文件
        DataSource<String> input = env.readTextFile("test.txt");
        //将读取到的文件举行输出
        input.print();
        //转换为DataSet
        DataSet<Orders> inputDataSet = input.map(new MapFunction<String, Orders>() {
            @Override
            public Orders map(String s) throws Exception {
                String[] splits = s.split(" ");
                return Orders.of(Integer.valueOf(splits[0]), String.valueOf(splits[1]), String.valueOf(splits[2]), Double.valueOf(splits[3]));
            }
        });
        //转换为table
        Table order = tableEnv.fromDataSet(inputDataSet);
        //注册Orders表名
        tableEnv.registerTable("Orders", order);
        Table nameResult = tableEnv.scan("Orders").select("name");
        //输出一下表
        nameResult.printSchema();

        //实行一下查询
        Table sqlQueryResult = tableEnv.sqlQuery("select name, sum(price) as total from Orders group by name order by total desc");
        //查询效果转换为DataSet
        DataSet<Result> result = tableEnv.toDataSet(sqlQueryResult, Result.class);
        result.print();

        //以tuple的体式格局举行输出
        result.map(new MapFunction<Result, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(Result result) throws Exception {
                String name = result.name;
                Double total = result.total;
                return Tuple2.of(name, total);
            }
        }).print();

        TableSink sink  = new CsvTableSink("SQLText.txt", " | ");

        //设置字段名
        String[] filedNames = {"name", "total"};
        //设置字段范例
        TypeInformation[] filedTypes = {Types.STRING(), Types.DOUBLE()};

        tableEnv.registerTableSink("SQLTEXT", filedNames, filedTypes, sink);

        sqlQueryResult.insertInto("SQLTEXT");

        env.execute();

    }

    public static class Orders {
        public Integer id;
        public String name;
        public String book;
        public Double price;

        public Orders() {
            super();
        }

        public static Orders of(Integer id, String name, String book, Double price) {
            Orders orders = new Orders();
            orders.id = id;
            orders.name = name;
            orders.book = book;
            orders.price = price;
            return orders;
        }
    }

    public static class Result {
        public String name;
        public Double total;

        public Result() {
            super();
        }

        public static Result of(String name, Double total) {
            Result result = new Result();
            result.name = name;
            result.total = total;
            return result;
        }
    }

}

Scala函数式编程（五） 函数式的错误处理