天天看點
傳回

Flink-之如何使用Table&SQL APIFlink-之如何使用Table&SQL API

Flink-之如何使用Table&SQL API

1 maven依賴

首先通常需要引入以下依賴。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.shufang</groupId>
    <artifactId>flink-demo-project-20210501</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- 1.9之前老的Old Table&SQL planner，這個依賴中已經包括了java‘scala的橋接依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>

        <!-- 1.9版本及之後引入的blink blanner<阿裡開源的~>，這個依賴中已經包括了java‘scala的橋接依賴-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>

        <!-- 1.9版本及之後引入的blink blanner runtime<阿裡開源的~> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime-blink_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>

        <!--TableSQL的javaAPI依賴-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!--TableSQL的scalaAPI依賴-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!--使用者自定義函數的相關依賴-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.10.1</version>
        </dependency>

        <dependency>
            <groupId>org.jetbrains</groupId>
            <artifactId>annotations</artifactId>
            <version>RELEASE</version>
            <scope>compile</scope>
        </dependency>

        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.13</version>
        </dependency>

    </dependencies>

</project>

2 如何建立Table&SQL API運作時環境

Table&SQL API運作時環境是程式的入口，與Spark中使用的裝飾着模式一樣，下面介紹如何使用不同的planner建立不同的TableEnvironment。

– TableEnvironment

​ – StreamTableEnvironment

​ – BatchTableEnvironment

package com.shufang.table_sql;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;

import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.ConnectorDescriptor;

/**
 * 本類講解如何使用JavaAPI調用Blink&Old planner建立對應的Table執行環境
 */
public class TableApiQuickStart_01 {
    public static void main(String[] args) {

        /*
         * 1.1 使用older planner接受流式資料源環境
         */
        //EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
        //StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);


        /*
         * 1.2 使用older planner接受批次資料源環境
         */
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment benv = BatchTableEnvironment.create(env);


        /*
         * 2.1 使用blink planner建構流式資料源環境
         */
        //EnvironmentSettings envSetting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, envSetting);


        /*
         * 2.2 使用blink planner建構批次資料源環境
         */
        //EnvironmentSettings envSetting = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
        //TableEnvironment tableEnv = TableEnvironment.create(envSetting);

    }
}

3 StreamTableEnvironment簡單嘗試 

package com.shufang.table_sql;


import com.shufang.beans.SensorTemper;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
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.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * 如何使用Java API完成以下過程
 * 1、注冊一個StreamTable
 * 2、查詢一個Table
 * 3、發射一個Table
 *
 * root
 *  |-- id: STRING
 *  |-- tempe: DOUBLE
 *
 * tableResult > sensor1,36.7
 * sqlResult > sensor1,36.7
 * tableResult > sensor1,34.1
 * sqlResult > sensor1,34.1
 * tableResult > sensor1,30.2
 * sqlResult > sensor1,30.2
 * sqlResult > sensor2,18.3
 * sqlResult > sensor2,36.1
 */
public class TableApiQuickStart_02 {
    public static void main(String[] args) throws Exception {

        // 1 建立執行環境，假設從檔案建立一個表,如果不指定panner，預設使用OldPlanner

        EnvironmentSettings setting = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,setting);



        // 2 從檔案建立一個DataStream
        DataStreamSource<String> fileStream = env.readTextFile("src/main/resources/sensor.txt");
        // 轉換成POJO類型
        SingleOutputStreamOperator<SensorTemper> sensorStream = fileStream.map(new MapFunction<String, SensorTemper>() {
            @Override
            public SensorTemper map(String s) throws Exception {
                String[] fields = s.split(",");
                return new SensorTemper(fields[0], new Double(fields[1]));
            }
        });


        // 将DataStream轉換成一個Table，并完成注冊
        Table table = tableEnv.fromDataStream(sensorStream);

        table.printSchema();

        // 3 查詢一個表
        // 3.1 使用table api進行查詢
        Table tableResult = table.select("id,tempe").where("id = 'sensor1'");
        tableEnv.createTemporaryView("sensor",table);
        String sql = "select id,tempe from sensor";
        Table sqlResult = tableEnv.sqlQuery(sql);


        // 4 分别列印不同的API的結果，首先轉換成DataStream
        tableEnv.toAppendStream(tableResult, Row.class).print("tableResult ");
        tableEnv.toAppendStream(sqlResult, Row.class).print("sqlResult ");


        // 5 最終使用env.execute()執行
        env.execute();
    }
}
Flink Table API sql api Flink SQL
上一篇: Chrome插件推薦一、CLEAN crxMouse Gestures二、LastPass: Free Password Manager三、下載下傳+四、微網誌圖床五、Octotree六、有道詞典Chrome劃詞插件七、Separate Window
下一篇: JPA開發總結&lt;一&gt;

繼續閱讀