4.2.5 Flink-流处理框架-Table API 与 SQL-表的输出

r囧r小猫 2022-11-07 13:09 252阅读 0赞

**目录**

1.写在前面

2.输出到文件

更新模式（Update Mode）

代码总结：

3.输出到 Kafka

代码总结

4.输出到 ElasticSearch

5.输出到 MySql

--------------------

# 1.写在前面 #

表的输出，是通过将数据写入 **TableSink** 来实现的。TableSink 是一个通用接口，可以 支持不同的文件格式、存储数据库和消息队列。具体实现，输出表最直接的方法，就是通过 Table.insertInto() 方法将一个 Table 写入注册过的 TableSink 中。

![20210312104027202.png][]

# 2.输出到文件 #

注册输出表

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

**insertInto（只实现了追加模式 append mode）**不支持数据聚合有更新操作的数据输出到文件。flink流处理过程中有个更新模式（Update Mode）的概念。

## 更新模式（Update Mode） ##

>         在流处理过程中，表的处理并不像传统定义的那样简单。 对于流式查询（Streaming Queries），需要声明如何在（动态）表和外部连接器之间执行转换。与外部系统交换的消息类型，由**更新模式（update mode）**指定。 Flink Table API 中的更新模式有以下三种：
> 
> **1）追加模式（Append Mode）：所有都支持**
> 
>         在追加模式下，表（动态表）和外部连接器只有**插入（Insert）消息**操作。
> 
> **2）撤回模式（Retract Mode）：**
> 
>         在撤回模式下，表和外部连接器交换的是：**添加（Add）和撤回（Retract）**消息。**更新一条数据需要发两条消息**。
> 
> ![20210312105155175.png][]
> 
> **3）Upsert（更新插入）模式：数据库要支持key**
> 
> **        更新一条数据只需要发一条消息，效率更高。**
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

## 代码总结：  ##

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.DataTypes;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.descriptors.Csv;
    import org.apache.flink.table.descriptors.FileSystem;
    import org.apache.flink.table.descriptors.Schema;
    
    public class TableTest3_FileOutput {
        public static void main(String[] args) throws Exception {
            // 1. 创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 2. 表的创建：连接外部系统，读取数据
            // 读取文件
            String filePath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt";
            tableEnv.connect( new FileSystem().path(filePath))
                    .withFormat( new Csv())
                    .withSchema( new Schema()
                            .field("id", DataTypes.STRING())
                            .field("timestamp", DataTypes.BIGINT())
                            .field("temp", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("inputTable");
    
            Table inputTable = tableEnv.from("inputTable");
    //        inputTable.printSchema();
    //        tableEnv.toAppendStream(inputTable, Row.class).print();
    
            // 3. 查询转换
            // 3.1 Table API
            // 简单转换
            Table resultTable = inputTable.select("id, temp")
                    .filter("id === 'sensor_6'");
    
            // 聚合统计
            Table aggTable = inputTable.groupBy("id")
                    .select("id, id.count as count, temp.avg as avgTemp");
    
            // 3.2 SQL
            tableEnv.sqlQuery("select id, temp from inputTable where id = 'senosr_6'");
            Table sqlAggTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id");
    
            // 4. 输出到文件
            // 连接外部文件注册输出表
            String outputPath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\out.txt";
            tableEnv.connect( new FileSystem().path(outputPath))
                    .withFormat( new Csv())
                    .withSchema( new Schema()
                            .field("id", DataTypes.STRING())
    //                        .field("cnt", DataTypes.BIGINT())
                            .field("temperature", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("outputTable");
    
            resultTable.insertInto("outputTable");
    //        aggTable.insertInto("outputTable");//聚合更新数据，没有办法插入到文件，会直接报错
    
            env.execute();
        }
    }

# 3.输出到 Kafka #

除了输出到文件，也可以输出到 Kafka。我们可以结合前面 Kafka 作为输入数据，构建数据管道，kafka 进，kafka 出。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

## 代码总结 ##

aggTable也是没有办法写入到kafka中，原因同上面的写入文件失败。因为不光有追加还有更改操作。适合日志操作。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.DataTypes;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.table.descriptors.Csv;
    import org.apache.flink.table.descriptors.Kafka;
    import org.apache.flink.table.descriptors.Schema;
    
    public class TableTest4_KafkaPipeLine {
        public static void main(String[] args) throws Exception {
            // 1. 创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            // 2. 连接Kafka，读取数据
            tableEnv.connect(new Kafka()
                    .version("0.11")
                    .topic("sensor")
                    .property("zookeeper.connect", "localhost:2181")
                    .property("bootstrap.servers", "localhost:9092")
            )
                    .withFormat(new Csv())
                    .withSchema(new Schema()
                            .field("id", DataTypes.STRING())
                            .field("timestamp", DataTypes.BIGINT())
                            .field("temp", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("inputTable");
    
            // 3. 查询转换
            // 简单转换
            Table sensorTable = tableEnv.from("inputTable");
            Table resultTable = sensorTable.select("id, temp")
                    .filter("id === 'sensor_6'");
    
            // 聚合统计
            Table aggTable = sensorTable.groupBy("id")
                    .select("id, id.count as count, temp.avg as avgTemp");
    
            // 4. 建立kafka连接，输出到不同的topic下
            tableEnv.connect(new Kafka()
                    .version("0.11")
                    .topic("sinktest")
                    .property("zookeeper.connect", "localhost:2181")
                    .property("bootstrap.servers", "localhost:9092")
            )
                    .withFormat(new Csv())
                    .withSchema(new Schema()
                            .field("id", DataTypes.STRING())
    //                        .field("timestamp", DataTypes.BIGINT())
                            .field("temp", DataTypes.DOUBLE())
                    )
                    .createTemporaryTable("outputTable");
    
            resultTable.insertInto("outputTable");
    
            env.execute();
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

# 4.输出到 ElasticSearch #

ElasticSearch 的 connector 可以在 **upsert（update+insert，更新插入）模式**下操作，这样 就可以使用 Query 定义的键（key）与外部系统交换 UPSERT/DELETE 消息。 另外，对于“仅追加”（append-only）的查询，connector 还可以在 append 模式下操作， 这样就可以与外部系统只交换 insert 消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4][]

# 5.输出到 MySql #

Flink 专门为 Table API 的 jdbc 连接提供了 flink-jdbc 连接器，我们需要先引入依赖：

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-jdbc_2.12</artifactId>
        <version>1.10.1</version>
    </dependency>

jdbc 连接的代码实现比较特殊，因为没有对应的 java/scala 类实现 ConnectorDescriptor， 所以不能直接tableEnv.connect()。不过Flink SQL留下了执行DDL的接口：**tableEnv.sqlUpdate()**。 对于 jdbc 的创建表操作，天生就适合直接写 DDL 来实现，所以我们的代码可以这样写：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5][]

[20210312104027202.png]: /images/20221023/812fd7174e7d4ab4836232048750e930.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/0804c951e54c483e953692960b98142b.png
[20210312105155175.png]: /images/20221023/9e2ae66a4ce74ce4979fa913f187eab9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/772fdc7367fb4a5ca37c79b3fe87fb01.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/cdfc4375174041a58b3fd0b3cbde5f69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221023/e04483675ccb44c49540ae7989a41eba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4]: /images/20221023/1a42f91305fa44f697b668e8f1dd80a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 5]: /images/20221023/b1c8c25360974143913e7c50d0c83002.png