Flink-之如何使用TableSQL API

本文主要是介绍Flink-之如何使用TableSQL 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，默认使用OldPlannerEnvironmentSettings setting = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,setting);// 2 从文件创建一个DataStreamDataStreamSource<String> fileStream = env.readTextFile("src/main/resources/sensor.txt");// 转换成POJO类型SingleOutputStreamOperator<SensorTemper> sensorStream = fileStream.map(new MapFunction<String, SensorTemper>() {@Overridepublic 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的结果，首先转换成DataStreamtableEnv.toAppendStream(tableResult, Row.class).print("tableResult ");tableEnv.toAppendStream(sqlResult, Row.class).print("sqlResult ");// 5 最终使用env.execute()执行env.execute();}
}