大数据开发工程师-第十六周 Flink极速上手篇-Flink核心API之TableAPI和SQL-5

发表于 | 分类于 ， | 浏览 次
字数统计: 4.1k | 阅读时长 ≈ 20

第十六周 Flink极速上手篇-Flink核心API之TableAPI和SQL-5

Table API & SQL

1
注意:Table API和SQL现在还处于活跃开发阶段，还没有完全实现Flink中所有的特性。不是所有的[Table API，SQL]和[流，批]的组合都是支持的。
1
2
3
Table API和SQL的由来：
Flink针对标准的流处理和批处理提供了两种关系型API，Table API和SQL。Table API允许用户以一种很直观的方式进行select、filter和join操作。Flink SQL基于Apache Calcite实现标准SQL。针对批处理和流处理可以提供相同的处理语义和结果。
Flink Table API、SQL和Flink的DataStream API、DataSet API是紧密联系在一起的。
1
2
3
Table API和SQL是一种关系型API，用户可以像操作Mysql数据库表一样的操作数据，而不需要写代码，更不需要手工的对代码进行调优。另外，SQL作为一个非程序员可操作的语言，学习成本很低，如果一个系统提供 SQL支持，将很容易被用户接受。

如果你想要使用Table API和SQL的话，需要添加下面的依赖
1
2
3
4
5
6
7
8
9
10
11
12
<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-table-api-java-bridge_2.12</artifactId>
     <version>1.11.0</version>
     <scope>provided</scope>
    </dependency>
<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
     <version>1.11.0</version>
     <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
如果你想在 本地 IDE中运行程序，还需要添加下面的依赖

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-table-planner-blink_2.12</artifactId>
     <version>1.11.0</version>
     <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
如果你用到了老的执行引擎，还需要添加下面这个依赖

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-table-planner_2.12</artifactId>
     <version>1.11.1</version>
     <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
注意：由于部分table相关的代码是用Scala实现的，所以，这个依赖也是必须的。【这个依赖我们在前面开发DataStream程序的时候已经添加过了】
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.12</artifactId>
    <version>1.11.0</version>
    <scope>provided</scope>
</dependency>
1
2
3
4
Table API和SQL通过join API集成在一起，这个join API的核心概念是Table，Table可以作为查询的输入和输出。
针对Table API和SQL我们主要讲解以下内容
1：Table API和SQL的使用
2：DataStream、DataSet和Table之间的互相转换

Table API 和SQL的使用

创建TableEnvironment对象

1
2
想要使用Table API和SQL，首先要创建一个TableEnvironment对象。
下面我们来创建一个TableEnvironment对象

scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package com.imooc.scala.tablesql
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.{BatchTableEnvironment, Stream
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
/**
 * 创建TableEnvironment对象
 * Created by xuwei
 */
object CreateTableEnvironmentScala {
     def main(args: Array[String]): Unit = {
         /**
         * 注意：如果Table API和SQL不需要和DataStream或者DataSet互相转换
         * 则针对stream和batch都可以使用TableEnvironment
         */
         
         //指定底层使用Blink引擎，以及数据处理模式-stream
         //从1.11版本开始，Blink引擎成为Table API和SQL的默认执行引擎，在生产环境下面，推
         val sSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
         //创建TableEnvironment对象
         val sTableEnv = TableEnvironment.create(sSettings)
         
         //指定底层使用Blink引擎，以及数据处理模式-batch
         val bSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
         //创建TableEnvironment对象
         val bTableEnv = TableEnvironment.create(bSettings)
         
         
         /**
         * 注意：如果Table API和SQL需要和DataStream或者DataSet互相转换
         * 针对stream需要使用StreamTableEnvironment
         * 针对batch需要使用BatchTableEnvironment
         */
         //创建StreamTableEnvironment
         val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
         val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
         val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
         
         //创建BatchTableEnvironment
         //注意：此时只能使用旧的执行引擎，新的Blink执行引擎不支持和DataSet转换
         val bbEnv = ExecutionEnvironment.getExecutionEnvironment
         val bbTableEnv = BatchTableEnvironment.create(bbEnv)
     }
}

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package com.imooc.java.tablesql;
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.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/**
 * 创建TableEnvironment对象
 * Created by xuwei
 */
public class CreateTableEnvironmentJava {
     public static void main(String[] args) {
         /**
         * 注意：如果Table API和SQL不需要和DataStream或者DataSet互相转换
         * 则针对stream和batch都可以使用TableEnvironment
         */
         //创建TableEnvironment对象-stream
         EnvironmentSettings sSettings = EnvironmentSettings.newInstance().use
         TableEnvironment sTableEnv = TableEnvironment.create(sSettings);
         //创建TableEnvironment对象-batch
         EnvironmentSettings bSettings = EnvironmentSettings.newInstance().use
         TableEnvironment bTableEnv = TableEnvironment.create(bSettings);
         /**
         * 注意：如果Table API和SQL需要和DataStream或者DataSet互相转换
         * 针对stream需要使用StreamTableEnvironment
         * 针对batch需要使用BatchTableEnvironment
         */
         //创建StreamTableEnvironment
         StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExec
         EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().us
         StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssE
         //创建BatchTableEnvironment
         //注意：此时只能使用旧的执行引擎，新的Blink执行引擎不支持和DataSet转换
         ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnviron
         BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv
     }
}

Table API和 SQL的使用

1
2
3
4
下面我们来演示一下Table API和SQL的使用
目前创建Table的很多方法都过时了，都不推荐使用了，例如：registerTableSource、connect等方法
目前官方推荐使用executeSql的方式，executeSql里面支持
1DDL/DML/DQL/SHOW/DESCRIBE/EXPLAIN/USE等语法

scala

image-20230412105247826

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package com.imooc.scala.tablesql
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
/**
 * TableAPI 和 SQL的使用
 * Created by xuwei
 */
object TableAPIAndSQLOpScala {
     def main(args: Array[String]): Unit = {
         //获取TableEnvironment
         val sSettings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode().build()
         val sTableEnv = TableEnvironment.create(sSettings)
         //创建输入表
         /**
         * connector.type：指定connector的类型
         * connector.path：指定文件或者目录地址
         * format.type：文件数据格式化类型
         * 注意：SQL语句如果出现了换行，行的末尾可以添加空格或者\n都可以，最后一行不用添
         */
         sTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")")
        
         //使用Table API实现数据查询和过滤等操作
         /*import org.apache.flink.table.api._
         val result = sTableEnv.from("myTable")
         .select($"id",$"name")
         .filter($"id" > 1)*/
        
         //使用SQL实现数据查询和过滤等操作
         val result = sTableEnv.sqlQuery("select id,name from myTable where id > 1" )
         //输出结果到控制台
         result.execute.print()
          //创建输出表
         sTableEnv.executeSql("" +
         "create table newTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\res',\n" +
         "'format.type' = 'csv'\n" +
         ")")
         //输出结果到表newTable中
         result.executeInsert("newTable")
     }
}

image-20230412105345232

1
2
3
4
5
6
7
8
9
10
11
12
13
注意：针对SQL建表语句的写法还有一种比较清晰的写法

sTableEnv.executeSql(
 """
 |create table myTable(
 |id int,
 |name string
 |) with (
 |'connector.type' = 'filesystem',
 |'connector.path' = 'D:\data\source',
 |'format.type' = 'csv'
 |)
 |""".stripMargin)

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
package com.imooc.java.tablesql;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import static org.apache.flink.table.api.Expressions.$;
/**
 * TableAPI 和 SQL的使用
 * Created by xuwei
 */
public class TableAPIAndSQLOpJava {
     public static void main(String[] args) {
         //获取TableEnvironment
         EnvironmentSettings sSettings = EnvironmentSettings.newInstance().use
         TableEnvironment sTableEnv = TableEnvironment.create(sSettings);
         //创建输入表
         sTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")");
         //使用Table API实现数据查询和过滤等操作
         /*Table result = sTableEnv.from("myTable")
         .select($("id"), $("name"))
         .filter($("id").isGreater(1));*/
             //使用SQL实现数据查询和过滤等操作
         Table result = sTableEnv.sqlQuery("select id,name from myTable where 
         //输出结果到控制台
         result.execute().print();
         //创建输出表
         sTableEnv.executeSql("" +
         "create table newTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\res',\n" +
         "'format.type' = 'csv'\n" +
         ")");
         //输出结果到表newTable中
         result.executeInsert("newTable");
     }
}

DataStream、DataSet和Table之间的互相转换

1
2
Table API和SQL可以很容易的和DataStream和DataSet程序集成到一块。通过TableEnvironment，可以把DataStream或者DataSet注册为Table，这样就可以使用Table API和SQL查询了。通过
TableEnvironment也可以把Table对象转换为DataStream或者DataSet，这样就可以使用DataStream或者DataSet中的相关API了。

使用DataStream创建表

1
2
3
1：使用DataStream创建表，主要包含下面这两种情况
使用DataStream创建view视图
使用DataStream创建table对象

scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com.imooc.scala.tablesql
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
/**
 * 将DataStream转换成表
 * Created by xuwei
 */
object DataStreamToTableScala {
     def main(args: Array[String]): Unit = {
         //获取StreamTableEnvironment
         val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
         val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
         val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
         
         //获取DataStream
         import org.apache.flink.api.scala._
         val stream = ssEnv.fromCollection(Array((1, "jack"), (2, "tom"), (3, "mack")))
         //第一种：将DataStream转换为view视图
         import org.apache.flink.table.api._
   ssTableEnv.createTemporaryView("myTable",stream,'id,'name)
         ssTableEnv.sqlQuery("select * from myTable where id > 1").execute().print()
         
         //第二种：将DataStream转换为table对象
         val table = ssTableEnv.fromDataStream(stream, $"id", $"name")
         table.select($"id",$"name")
         .filter($"id" > 1)
         .execute()
         .print()
         //注意：'id,'name 和 $"id", $"name" 这两种写法是一样的效果
     }
}

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package com.imooc.java.tablesql;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.bridge.java.StreamTableEnvironment;
import java.util.ArrayList;
import static org.apache.flink.table.api.Expressions.$;
/**
 * 将DataStream转换成表
 * Created by xuwei
 */
public class DataStreamToTableJava {
     public static void main(String[] args) {
         //获取StreamTableEnvironment
         StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExec
         EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().us
         StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssE
         //获取DataStream
         ArrayList<Tuple2<Integer, String>> data = new ArrayList<>();
         data.add(new Tuple2<Integer,String>(1,"jack"));
         data.add(new Tuple2<Integer,String>(2,"tom"));
         data.add(new Tuple2<Integer,String>(3,"mick"));
         DataStreamSource<Tuple2<Integer, String>> stream = ssEnv.fromCollecti
         //第一种：将DataStream转换为view视图
         ssTableEnv.createTemporaryView("myTable",stream,$("id"),$("name"));
         ssTableEnv.sqlQuery("select * from myTable where id > 1").execute().p
         //第二种：将DataStream转换为table对象
         Table table = ssTableEnv.fromDataStream(stream, $("id"), $("name"));
         table.select($("id"), $("name"))
         .filter($("id").isGreater(1))
         .execute()
         .print();
    }
}

使用DataSet创建表

1
注意：此时只能使用旧的执行引擎，新的Blink执行引擎不支持和DataSet转换

scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com.imooc.scala.tablesql
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.{BatchTableEnvironment, Stream
/**
 * 将DataSet转换成表
 * Created by xuwei
 */
object DataSetToTableScala {
     def main(args: Array[String]): Unit = {
         //获取BatchTableEnvironment
         val bbEnv = ExecutionEnvironment.getExecutionEnvironment
         val bbTableEnv = BatchTableEnvironment.create(bbEnv)
        
         //获取DataSet
         import org.apache.flink.api.scala._
         val set = bbEnv.fromCollection(Array((1, "jack"), (2, "tom"), (3, "mack")))
         //第一种：将DataSet转换为view视图
         import org.apache.flink.table.api._
         bbTableEnv.createTemporaryView("myTable",set,'id,'name)
         bbTableEnv.sqlQuery("select * from myTable where id > 1").execute().print
         
         //第二种：将DataSet转换为table对象
         val table = bbTableEnv.fromDataSet(set, $"id", $"name")
         table.select($"id",$"name")
         .filter($"id" > 1)
         .execute()
         .print()
         //注意：'id,'name 和 $"id", $"name" 这两种写法是一样的效果
     }
}

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
package com.imooc.java.tablesql;
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.streaming.api.datastream.DataStreamSource;
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.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.util.ArrayList;
import static org.apache.flink.table.api.Expressions.$;
/**
 * 将DataSet转换成表
 * Created by xuwei
 */
public class DataSetToTableJava {
     public static void main(String[] args) {
         //获取BatchTableEnvironment
         ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnviron
         BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv
         //获取DataSet
         ArrayList<Tuple2<Integer, String>> data = new ArrayList<>();
         data.add(new Tuple2<Integer,String>(1,"jack"));
         data.add(new Tuple2<Integer,String>(2,"tom"));
         data.add(new Tuple2<Integer,String>(3,"mick"));
         DataSource<Tuple2<Integer, String>> set = bbEnv.fromCollection(data);
         //第一种：将DataSet转换为view视图
         bbTableEnv.createTemporaryView("myTable",set,$("id"),$("name"));
         bbTableEnv.sqlQuery("select * from myTable where id > 1").execute().p
         //第二种：将DataSet转换为table对象
         Table table = bbTableEnv.fromDataSet(set, $("id"), $("name"));
         table.select($("id"), $("name"))
         .filter($("id").isGreater(1))
         .execute()
         .print();
     }
}
1
2
3
4
5
6
将Table转换为DataStream或者DataSet时，你需要指定生成的 DataStream或者DataSet的数据类型，即，Table 的每行数据要转换成的数据类型。通常最方便的选择是转换成Row。以下列表概述了不同选项的功能：
Row: 通过角标映射字段，支持任意数量的字段，支持null值，无类型安全（type-safe）检查。
POJO: Java中的实体类，这个实体类中的字段名称需要和Table中的字段名称保持一致，支持任意数量的字段，支持null值，有类型安全检查。
Case Class: 通过角标映射字段，不支持null值，有类型安全检查。
Tuple: 通过角标映射字段，Scala中限制22个字段，Java中限制25个字段，不支持null值，有类型安全检查。
Atomic Type: Table必须有一个字段，不支持null值，有类型安全检查。

将Table转换为DataStream

1
2
3
4
5
3：将表转换成DataStream流式查询的结果Table会被动态地更新，即每个新的记录到达输入流时结果就会发生变化。因此，转换此动态查询的DataStream需要对表的更新进行编码。

有几种模式可以将Table转换为DataStream。
Append Mode:这种模式只适用于当动态表仅由INSERT更改修改时(仅附加)，之前添加的数据不会被更新。
Retract Mode:可以始终使用此模式，它使用一个Boolean标识来编码INSERT和DELETE更改。

scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
package com.imooc.scala.tablesql
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row
/**
 * 将table转换成 DataStream
 * Created by xuwei
 */
object TableToDataStreamScala {
     def main(args: Array[String]): Unit = {
         //获取StreamTableEnvironment
         val ssEnv = StreamExecutionEnvironment.getExecutionEnvironment
         val ssSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inSt
         val ssTableEnv = StreamTableEnvironment.create(ssEnv, ssSettings)
         //创建输入表
         ssTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")")
         //获取table
         val table = ssTableEnv.from("myTable")
         //将table转换为DataStream
         //如果只有新增(追加)操作，可以使用toAppendStream
         import org.apache.flink.api.scala._
         val appStream = ssTableEnv.toAppendStream[Row](table)
         appStream.map(row=>(row.getField(0).toString.toInt,row.getField(1).toString))
         .print()
         
         //如果有增加操作，还有删除操作，则使用toRetractStream
         val retStream = ssTableEnv.toRetractStream[Row](table)
         retStream.map(tup=>{
         val flag = tup._1
         val row = tup._2
         val id = row.getField(0).toString.toInt
         val name = row.getField(1).toString
         (flag,id,name)
         }).print()
         //注意：将table对象转换为DataStream之后，就需要调用StreamExecutionEnvironment
         ssEnv.execute("TableToDataStreamScala")
     }
}

image-20230412115003343

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
package com.imooc.java.tablesql;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
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.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
/**
 * 将table转换成 DataStream
 * Created by xuwei
 */
public class TableToDataStreamJava {
     public static void main(String[] args) throws Exception{
         //获取StreamTableEnvironment
         StreamExecutionEnvironment ssEnv = StreamExecutionEnvironment.getExec
         EnvironmentSettings ssSettings = EnvironmentSettings.newInstance().us
         StreamTableEnvironment ssTableEnv = StreamTableEnvironment.create(ssE
         //创建输入表
         ssTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")");
         //获取table
         Table table = ssTableEnv.from("myTable");
         //将table转换为DataStream
         //如果只有新增(追加)操作，可以使用toAppendStream
                                                                 DataStream<Row> appStream = ssTableEnv.toAppendStream(table, Row.clas
         appStream.map(new MapFunction<Row, Tuple2<Integer,String>>() {
         @Override
         public Tuple2<Integer, String> map(Row row)
         throws Exception {
         int id = Integer.parseInt(row.getField(0).toString());
         String name = row.getField(1).toString();
         return new Tuple2<Integer, String>(id, name);
         }
         }).print();
         //如果有增加操作，还有删除操作，则使用toRetractStream
         DataStream<Tuple2<Boolean, Row>> retStream = ssTableEnv.toRetractStre
         retStream.map(new MapFunction<Tuple2<Boolean, Row>, Tuple3<Boolean,Int
         @Override
         public Tuple3<Boolean, Integer, String> map(Tuple2<Boolean, Row> t
         throws Exception {
         Boolean flag = tup.f0;
         int id = Integer.parseInt(tup.f1.getField(0).toString());
         String name = tup.f1.getField(1).toString();
         return new Tuple3<Boolean, Integer, String>(flag, id, name);
         }
         }).print();
         ssEnv.execute("TableToDataStreamJava");
 }

将表转换成DataSet

scala

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package com.imooc.scala.tablesql
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.BatchTableEnvironment
import org.apache.flink.types.Row
/**
 * 将table转换成 DataSet
 * Created by xuwei
 */
object TableToDataSetScala {
     def main(args: Array[String]): Unit = {
         //获取BatchTableEnvironment
         val bbEnv = ExecutionEnvironment.getExecutionEnvironment
         val bbTableEnv = BatchTableEnvironment.create(bbEnv)
         //创建输入表
         bbTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")")
         //获取table
         val table = bbTableEnv.from("myTable")
         //将table转换为DataSet
         import org.apache.flink.api.scala._
         val set = bbTableEnv.toDataSet[Row](table)
         set.map(row=>(row.getField(0).toString.toInt,row.getField(1).toString))
         .print()
     }
}

image-20230412115320971

java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
package com.imooc.java.tablesql;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
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.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
/**
* 将table转换成 DataSet
 * Created by xuwei
 */
public class TableToDataSetJava {
     public static void main(String[] args) throws Exception{
         //获取BatchTableEnvironment
         ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnviron
         BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv
         //创建输入表
         bbTableEnv.executeSql("" +
         "create table myTable(\n" +
         "id int,\n" +
         "name string\n" +
         ") with (\n" +
         "'connector.type' = 'filesystem',\n" +
         "'connector.path' = 'D:\\data\\source',\n" +
         "'format.type' = 'csv'\n" +
         ")");
         //获取table
         Table table = bbTableEnv.from("myTable");
         //将table转换为DataSet
         DataSet<Row> set = bbTableEnv.toDataSet(table, Row.class);
         set.map(new MapFunction<Row, Tuple2<Integer,String>>() {
         @Override
         public Tuple2<Integer, String> map(Row row)
         throws Exception {
         int id = Integer.parseInt(row.getField(0).toString());
         String name = row.getField(1).toString();
         return new Tuple2<Integer, String>(id, name);
         }
         }).print();
     }
}

本文标题:大数据开发工程师-第十六周 Flink极速上手篇-Flink核心API之TableAPI和SQL-5

文章作者:TTYONG

发布时间:2023年04月08日 - 23:04

最后更新:2023年06月02日 - 18:06

原始链接:http://tianyong.fun/%E5%A4%A7%E6%95%B0%E6%8D%AE%E5%BC%80%E5%8F%91%E5%B7%A5%E7%A8%8B%E5%B8%88-%E7%AC%AC%E5%8D%81%E5%85%AD%E5%91%A8-Flink%E6%9E%81%E9%80%9F%E4%B8%8A%E6%89%8B%E7%AF%87-Flink%E6%A0%B8%E5%BF%83API%E4%B9%8BTableAPI%E5%92%8CSQL-5.html

许可协议: 转载请保留原文链接及作者。

多少都是爱
0%