首页 > 其他分享 >Flink-Table API

Flink-Table API

时间：2022-09-29 22:57:39浏览次数：41  
标签：Flink val tableEnv flink API SQL Table

  在 Flink 提供的多层级 API 中，核心是 DataStream API，这是我们开发流处理应用的基本途径；底层则是所谓的处理函数（process function），可以访问事件的时间信息、注册定时器、自定义状态，进行有状态的流处理。DataStream API 和处理函数比较通用，有了这些 API，理论上我们就可以实现所有场景的需求了。 不过在企业实际应用中，往往会面对大量类似的处理逻辑，所以一般会将底层 API 包装成更加具体的应用级接口。怎样的接口风格最容易让大家接收呢？作为大数据工程师，我们最为熟悉的数据统计方式，当然就是写 SQL 了。 SQL 是结构化查询语言（Structured Query Language）的缩写，是我们对关系型数据库进行查询和修改的通用编程语言。在关系型数据库中，数据是以表（table）的形式组织起来的，所以也可以认为 SQL 是用来对表进行处理的工具语言。无论是传统架构中进行数据存储的MySQL、PostgreSQL，还是大数据应用中的 Hive，都少不了 SQL 的身影；而 Spark 作为大数据处理引擎，为了更好地支持在 Hive 中的 SQL 查询，也提供了 Spark SQL 作为入口。 Flink 同样提供了对于“表”处理的支持，这就是更高层级的应用 API，在 Flink 中被称为Table API 和 SQL。Table API 顾名思义，就是基于“表”（Table）的一套 API，它是内嵌在 Java、Scala 等语言中的一种声明式领域特定语言（DSL），也就是专门为处理表而设计的；在此基础上，Flink 还基于 Apache Calcite 实现了对 SQL 的支持。这样一来，我们就可以在 Flink 程序中直接写 SQL 来实现处理需求了。   在 Flink 中这两种 API 被集成在一起，SQL 执行的对象也是 Flink 中的表（Table），所以我们一般会认为它们是一体的。Flink 是批流统一的处理框架，无论是批处理（DataSet API）还是流处理（DataStream API），在上层应用中都可以直接使用 TableAPI 或者 SQL 来实现；这两种 API 对于一张表执行相同的查询操作，得到的结果是完全一样的。 需要说明的是，Table API 和 SQL 最初并不完善，在 Flink 1.9 版本合并阿里巴巴内部版本Blink 之后发生了非常大的改变，此后也一直处在快速开发和完善的过程中，直到 Flink 1.12版本才基本上做到了功能上的完善。而即使是在目前最新的 1.13 版本中，Table API 和 SQL 也依然不算稳定，接口用法还在不停调整和更新。所以这部分希望大家重在理解原理和基本用法，具体的 API 调用可以随时关注官网的更新变化。  

1.Table API的简单使用

  
package com.zhen.flink.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.Expressions.$

/**
  * @Author FengZhen
  * @Date 9/28/22 10:38 PM
  * @Description TODO
  */

case class Event(user: String, url: String, timeLength: Long)


object SimpleTableExample {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 读取数据源，创建DataStream
    val eventStream = env.fromElements(
      Event("Alice", "./home", 1000L),
      Event("Bob", "./cart", 1000L),
      Event("Alice", "./prod?id=1", 5 * 1000L),
      Event("Cary", "./home", 60 * 1000L),
      Event("Bob", "./prod?id=3", 90 * 1000L),
      Event("Alice", "./prod?id=7", 105 * 1000L),
    )

    // 创建表环境
    val tableEnv = StreamTableEnvironment.create(env)

    // 将DataStream转换成表
    val eventTable: Table = tableEnv.fromDataStream(eventStream)

    // 调用Table API进行转换计算
    /**
      * 这里的$符号是 Table API 中定义的“表达式”类 Expressions 中的一个静态方法，传入一
      * 个字段名称，就可以指代数据中对应字段，这个方法需要使用如下的方式进行手动导入。
      * import org.apache.flink.table.api.Expressions.$
      */
    val resultTable = eventTable.select($("user"), $("url"))
      .where($("user").isEqual("Alice"))

    // 直接写SQL SQL后直接 + eventTable，会自动注册一张和变量名同名的表
    val resultSqlTable = tableEnv.sqlQuery("select user, url from " + eventTable + " where user = 'Bob'")

    // 转换成流打印输出
    resultTable.printSchema()

    /**
      * resultStream> +I[Alice, ./home]
      * resultSqlStream> +I[Bob, ./cart]
      * resultStream> +I[Alice, ./prod?id=1]
      * resultSqlStream> +I[Bob, ./prod?id=3]
      * resultStream> +I[Alice, ./prod?id=7]
      * +I:这是表示每条数据都是“插入”（Insert）到表中的新增数据。
      */
    val resultStream = tableEnv.toDataStream(resultTable)
    resultStream.print("resultStream")

    val resultSqlStream = tableEnv.toDataStream(resultSqlTable)
    resultSqlStream.print("resultSqlStream")

    env.execute("simple table example")
  }

}

 

2.Table API代码整体架构

在 Flink 中，Table API 和 SQL 可以看作联结在一起的一套 API，这套 API 的核心概念就是“表”（Table）。在我们的程序中，输入数据可以定义成一张表；然后对这张表进行查询，就可以得到新的表，这相当于就是流数据的转换操作；最后还可以定义一张用于输出的表，负责将处理结果写入到外部系统。 我们可以看到，程序的整体处理流程与 DataStream API 非常相似，也可以分为读取数据源（Source）、转换（Transformation）、输出（Sink）三部分；只不过这里的输入输出操作不需要额外定义，只需要将用于输入和输出的表定义出来，然后进行转换查询就可以了。   程序基本架构如下： 
// 创建表环境
val tableEnv = ...;
// 创建输入表，连接外部系统读取数据
tableEnv.executeSql("CREATE TEMPORARY TABLE inputTable ... WITH ( 'connector'= ...)")
// 注册一个表，连接到外部系统，用于输出
tableEnv.executeSql("CREATE TEMPORARY TABLE outputTable ... WITH ( 'connector'= ...)")
// 执行 SQL 对表进行查询转换，得到一个新的表
val table1 = tableEnv.sqlQuery("SELECT ... FROM inputTable... ")
// 使用 Table API 对表进行查询转换，得到一个新的表
val table2 = tableEnv.from("inputTable").select(...)
// 将得到的结果写入输出表
val tableResult = table1.executeInsert("outputTable")
  这里不是从一个 DataStream 转换成 Table，而是通过执行 DDL（DataDefinition Language，数据定义语言）来直接创建一个表。这里执行的 CREATE 语句中用 WITH指定了外部系统的连接器，于是就可以连接外部系统读取数据了。这其实是更加一般化的程序架构，因为这样我们就可以完全抛开 DataStream API，直接用 SQL 语句实现全部的流处理过程。 而后面对于输出表的定义是完全一样的。可以发现，在创建表的过程中，其实并不区分“输入”还是“输出”，只需要将这个表“注册”进来、连接到外部系统就可以了；这里的 inputTable、outputTable 只是注册的表名，并不代表处理逻辑，可以随意更换。至于表的具体作用，则要等到执行后面的查询转换操作时才能明确。我们直接从 inputTable 中查询数据，那么 inputTable就是输入表；而 outputTable 会接收查询的结果进行写入，那么就是输出表。 在早期的版本中，有专门的用于输入输出的 TableSource 和 TableSink，这与流处理里的概念是一一对应的；不过这种方式与关系型表和 SQL 的使用习惯不符，所以已被弃用，不再区分 Source 和 Sink。  

3.创建表环境

对于 Flink 这样的流处理框架来说，数据流和表在结构上还是有所区别的。所以使用 TableAPI 和 SQL 需要一个特别的运行时环境，这就是所谓的“表环境”（TableEnvironment）。它主要负责： （1）注册 Catalog 和表； （2）执行 SQL 查询； （3）注册用户自定义函数（UDF）； （4）DataStream 和表之间的转换。 这里的 Catalog 就是“目录”，与标准 SQL 中的概念是一致的，主要用来管理所有数据库（database）和表（table）的元数据（metadata）。通过 Catalog 可以方便地对数据库和表进行查询的管理，所以可以认为我们所定义的表都会“挂靠”在某个目录下，这样就可以快速检索。在表环境中可以由用户自定义 Catalog，并在其中注册表和自定义函数（UDF）。默认的 Catalog就叫作 default_catalog。 每个表和 SQL 的执行，都必须绑定在一个表环境（TableEnvironment）中。TableEnvironment是 Table API 中提供的基本接口类，可以通过调用静态的 create()方法来创建一个表环境实例。方法需要传入一个环境的配置参数 EnvironmentSettings，它可以指定当前表环境的执行模式和计划器（planner）。执行模式有批处理和流处理两种选择，默认是流处理模式；计划器默认使用 blink planner。  

4.创建表

表（Table）是我们非常熟悉的一个概念，它是关系型数据库中数据存储的基本形式，也是 SQL 执行的基本对象。Flink 中的表概念也并不特殊，是由多个“行”数据构成的，每个行（Row）又可以有定义好的多个列（Column）字段；整体来看，表就是固定类型的数据组成的二维矩阵。 为了方便地查询表，表环境中会维护一个目录（Catalog）和表的对应关系。所以表都是通过 Catalog 来进行注册创建的。表在环境中有一个唯一的 ID，由三部分组成：目录（catalog）名，数据库（database）名，以及表名。在默认情况下，目录名为 default_catalog，数据库名为default_database。所以如果我们直接创建一个叫作 MyTable 的表，它的 ID 就是： default_catalog.default_database.MyTable 具体创建表的方式，有通过连接器（connector）和虚拟表（virtual tables）两种。

4.1连接器表（Connector Tables）

最直观的创建表的方式，就是通过连接器（connector）连接到一个外部系统，然后定义出对应的表结构。例如我们可以连接到 Kafka 或者文件系统，将存储在这些外部系统的数据以“表”的形式定义出来，这样对表的读写就可以通过连接器转换成对外部系统的读写了。当我们在表环境中读取这张表，连接器就会从外部系统读取数据并进行转换；而当我们向这张表写入数据，连接器就会将数据输出（Sink）到外部系统中。 在代码中，我们可以调用表环境的 executeSql()方法，可以传入一个 DDL 作为参数执行SQL 操作。这里我们传入一个 CREATE 语句进行表的创建，并通过 WITH 关键字指定连接到外部系统的连接器： tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable ... WITH ( 'connector'= ... )") 这里的 TEMPORARY 关键字可以省略 这里没有定义 Catalog 和 Database ， 所 以 都 是 默 认 的 ， 表 的 完 整 ID 就 是default_catalog.default_database.MyTable。如果希望使用自定义的目录名和库名，可以在环境中进行设置： tEnv.useCatalog("custom_catalog") tEnv.useDatabase("custom_database") 这样我们创建的表完整 ID 就变成了 custom_catalog.custom_database.MyTable。之后在表环境中创建的所有表，ID 也会都以 custom_catalog.custom_database 作为前缀。  

4.2虚拟表（Virtual Tables）

在环境中注册之后，我们就可以在 SQL 中直接使用这张表进行查询转换了。 val newTable = tableEnv.sqlQuery("SELECT ... FROM MyTable... ") 这里调用了表环境的 sqlQuery()方法，直接传入一条 SQL 语句作为参数执行查询，得到的结果是一个 Table 对象。Table 是 Table API 中提供的核心接口类，就代表了一个 Java 中定义的表实例。 得到的 newTable 是一个中间转换结果，如果之后又希望直接使用这个表执行 SQL，又该怎么做呢？由于 newTable 是一个 Table 对象，并没有在表环境中注册；所以我们还需要将这个中间结果表注册到环境中，才能在 SQL 中使用： tableEnv.createTemporaryView("NewTable", newTable) 注意：这里的第一个参数"NewTable"是注册的表名，而第二个参数 newTable 是 Java 中的Table 对象。 我们发现，这里的注册其实是创建了一个“虚拟表”（Virtual Table）。这个概念与 SQL 语法中的视图（View）非常类似，所以调用的方法也叫作创建“虚拟视图”（createTemporaryView）。视图之所以是“虚拟”的，是因为我们并不会直接保存这个表的内容，并没有“实体”；只是在用到这张表的时候，会将它对应的查询语句嵌入到 SQL 中。 注册为虚拟表之后，我们就又可以在 SQL 中直接使用 NewTable 进行查询转换了。不难看到，通过虚拟表可以非常方便地让 SQL 分步骤执行得到中间结果，这为代码编写提供了很大的便利。 另外，虚拟表也可以让我们在 Table API 和 SQL 之间进行自由切换。一个 Java 中的 Table对象可以直接调用 Table API 中定义好的查询转换方法，得到一个中间结果表；这跟对注册好的表直接执行 SQL 结果是一样的。  

5.表的查询

创建好了表，接下来自然就是对表进行查询转换了。对一个表的查询（Query）操作，就对应着流数据的转换（Transformation）处理。 Flink 为我们提供了两种查询方式：SQL 和 Table API。

5.1执行SQL进行查询

基于表执行 SQL 语句，是我们最为熟悉的查询方式。Flink 基于 Apache Calcite 来提供对SQL 的支持，Calcite 是一个为不同的计算平台提供标准 SQL 查询的底层工具，很多大数据框架比如 Apache Hive、Apache Kylin 中的 SQL 支持都是通过集成 Calcite 来实现的。 在代码中，我们只要调用表环境的 sqlQuery()方法，传入一个字符串形式的 SQL 查询语句就可以了。执行得到的结果，是一个 Table 对象。 
// 创建表环境
val tableEnv = ...
// 创建表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )")
// 查询用户 Alice 的点击事件，并提取表中前两个字段
val aliceVisitTable = tableEnv.sqlQuery(
"SELECT user, url " +
"FROM EventTable " +
"WHERE user = 'Alice' "
)
目前 Flink 支持标准 SQL 中的绝大部分用法，并提供了丰富的计算函数。这样我们就可以把已有的技术迁移过来，像在 MySQL、Hive 中那样直接通过编写 SQL 实现自己的处理需求，从而大大降低了 Flink 上手的难度。 例如，我们也可以通过 GROUP BY 关键字定义分组聚合，调用 COUNT()、SUM()这样的函数来进行统计计算： 
val urlCountTable = tableEnv.sqlQuery(
"SELECT user, COUNT(url) " +
"FROM EventTable " +
"GROUP BY user "
)
上面的例子得到的是一个新的 Table 对象，我们可以再次将它注册为虚拟表继续在 SQL中调用。另外，我们也可以直接将查询的结果写入到已经注册的表中，这需要调用表环境的executeSql()方法来执行 DDL，传入的是一个 INSERT 语句： 
// 注册表
tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )")
tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )")
// 将查询结果输出到 OutputTable 中
tableEnv.executeSql (
"INSERT INTO OutputTable " +
"SELECT user, url " +
"FROM EventTable " +
"WHERE user = 'Alice' "
)
 

5.2 调用Table API进行查询

另外一种查询方式就是调用 Table API。这是嵌入在 Java 和 Scala 语言内的查询 API，核心就是 Table 接口类，通过一步步链式调用 Table 的方法，就可以定义出所有的查询转换操作。每一步方法调用的返回结果，都是一个 Table。 由于Table API是基于Table的Java实例进行调用的，因此我们首先要得到表的Java对象。基于环境中已注册的表，可以通过表环境的 from()方法非常容易地得到一个 Table 对象： val eventTable = tableEnv.from("EventTable") 传入的参数就是注册好的表名。注意这里 eventTable 是一个 Table 对象，而 EventTable 是在环境中注册的表名。得到 Table 对象之后，就可以调用 API 进行各种转换操作了，得到的是一个新的 Table 对象： 
val maryClickTable = eventTable
.where($("user").isEqual("Alice"))
.select($("url"), $("user"))
这里每个方法的参数都是一个“表达式”（Expression），用方法调用的形式直观地说明了想要表达的内容；“$”符号用来指定表中的一个字段。上面的代码和直接执行 SQL 是等效的。 Table API 是嵌入编程语言中的 DSL，SQL 中的很多特性和功能必须要有对应的实现才可以使用，因此跟直接写 SQL 比起来肯定就要麻烦一些。目前 Table API 支持的功能相对更少，可以预见未来 Flink 社区也会以扩展 SQL 为主，为大家提供更加通用的接口方式；  

6.输出表

表的创建和查询，就对应着流处理中的读取数据源（Source）和转换（Transform）；而最后一个步骤 Sink，也就是将结果数据输出到外部系统，就对应着表的输出操作。 在代码上，输出一张表最直接的方法，就是调用 Table 的方法 executeInsert()方法将一个Table 写入到注册过的表中，方法传入的参数就是注册的表名。 
// 注册表，用于输出数据到外部系统
tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )")
// 经过查询转换，得到结果表
val result = ...
// 将结果表写入已注册的输出表中
result.executeInsert("OutputTable")
在底层，表的输出是通过将数据写入到 TableSink 来实现的。TableSink 是 Table API 中提供的一个向外部系统写入数据的通用接口，可以支持不同的文件格式（比如 CSV、Parquet）、存储数据库（比如 JDBC、HBase、Elasticsearch）和消息队列（比如 Kafka）。它有些类似于DataStream API 中调用 addSink()方法时传入的 SinkFunction，有不同的连接器对它进行了实现。   
package com.zhen.flink.table

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.Expressions.$


/**
  * @Author FengZhen
  * @Date 9/28/22 11:19 PM
  * @Description 通用API
  */
object CommonApiTest {

  def main(args: Array[String]): Unit = {

    // 1.创建表环境
    // 1.1 直接基于流执行环境创建
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val tableEnv = StreamTableEnvironment.create(env)

    // 1.2 传入一个环境的配置参数创建
    val settings = EnvironmentSettings.newInstance()
        .inStreamingMode()
        .useBlinkPlanner()
        .build()
    val tableEnvironment = TableEnvironment.create(settings)

    // 2.创建表
    tableEnv.executeSql("CREATE TABLE eventTable (" +
      " uid STRING," +
      " url STRING," +
      " ts BIGINT" +
      ") WITH (" +
      " 'connector' = 'filesystem'," +
      " 'path' = '/Users/FengZhen/Desktop/accumulate/0_project/flink_learn/src/main/resources/data/input/clicks.txt', " +
      " 'format' = 'csv' " +
      ")")

    // 3.表的查询转换
    // 3.1 SQL
    val resultTable = tableEnv.sqlQuery("select uid, url, ts from eventTable where uid = 'Alice'")
    // 统计每个用户访问频次
    val urlCountTable = tableEnv.sqlQuery("select uid, count(url) from eventTable group by uid")
    // 创建虚拟表
    tableEnv.createTemporaryView("tempTable", resultTable)


    // 3.2 Table API
    val eventTable = tableEnv.from("eventTable")
    val resultTable2 = eventTable
      .select($("url"), $("uid"), $("ts"))
      .where($("url").isEqual("./home"))


    // 4.输出表的创建
    val outputTable = tableEnv.executeSql("CREATE TABLE outputTable (" +
      " user_name STRING," +
      " url STRING," +
      " `timestamp` BIGINT" +
      ") WITH (" +
      " 'connector' = 'filesystem'," +
      " 'path' = '/Users/FengZhen/Desktop/accumulate/0_project/flink_learn/src/main/resources/data/output', " +
      " 'format' = 'csv' " +
      ")")

    // 5.将结果表写入输出表汇总
    resultTable.executeInsert("outputTable")
  }

}
pom.xml 
<?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.zhen.flink</groupId>
    <artifactId>flink_learn</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>flink_learn Maven</name>


    <properties>
        <scala_version>2.12</scala_version>
        <flink_version>1.13.1</flink_version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.bahir/flink-connector-redis -->
        <dependency>
            <groupId>org.apache.bahir</groupId>
            <artifactId>flink-connector-redis_2.11</artifactId>
            <version>1.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch6_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.44</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-statebackend-rocksdb -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <!--  Scala 的“桥接器”（bridge），主要就是负责 Table API 和下层 DataStreamAPI 的连接支持    -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <!--  “计划器”（planner），它是 Table API 的核心组件，负责提供运行时环境，并生成程序的执行计划。这里我们用到的是新版的 blink planner。由于 Flink 安装包的 lib 目录下会自带 planner，所以在生产集群环境中提交的作业不需要打包这个依赖      -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala_version}</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <!--  想实现自定义的数据格式来做序列化  -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink_version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>${flink_version}</version>
        </dependency>



    </dependencies>

    <build>
        <plugins> <!-- 该插件用于将 Scala 代码编译成 class 文件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.4.6</version>
                <executions>
                    <execution> <!-- 声明绑定到 maven 的 compile 阶段 -->
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>


</project>

 

 

标签：Flink,val,tableEnv,flink,API,SQL,Table
From： https://www.cnblogs.com/EnzoDin/p/16743404.html

相关文章

  • drf请求与相应(Request，Response)，drf能够解析的请求编码，响应编码，GenericAPIView和APIVi
    drf请求与响应Request类(请求)Response类(响应)drf 能够解析的请求编码，响应编码能够解析的请求编码响应编码GenericAPIView和APIView(2个视图基......
  • 前端面试总结12-WebApi-存储
    简述cooki，localstorage，sessionstorage的区别（1：cookie数据存放在浏览器上，session存放在服务器上（2：cookie安全性低（3：session占用服务器性能（4：单个cookie最大存储数据不超过4k......
  • Latex表格太宽怎么办？用 sidewaystable ！
    \begin{sidewaystable}[!tbp]\caption{ExampleofSideWaysTable}\centering\begin{tabular}{c|c|c|c|c|c|c|c|c|c|c|c|c}\hlineAAAAA&......
  • STRAPI概览
     总体介绍 实现所有内容多个多端重复使用。通过UI快速搭建数据库和API，节省大量后端工作量和学习成本。技术细节不用手动建立API路由。支持几乎所有常见的sql......
  • 前端面试总结11-WebApi-Ajax
    1.同源策略：ajax请求时，浏览器要求当前网页和serve必须同源（安全），即协议，域名，端口三者必须一致2.可无视同源策略的情况（1：<img/>可用于统计打点，可使用第三方统计服务（2：<link/><......
  • python使用win32api进行后台窗口的部分截图函数
    defwindow_capture_beat(hwnd,stayx:int,endx:int,stay:int,endy:int):hwndDC=win32gui.GetWindowDC(hwnd)mfcDC=win32ui.CreateDCFromHandle(hwndDC)......
  • element-ui v-table 复选框默认选中
    <el-tableref="refTable":data="list"v-loading="listLoading"element-loading-text="加载中"fithighlight-current-ro......
  • Editable FreeViewpoint Video
    使用分层神经表示的可编辑自由视点视频自由视点视频的生成对于沉浸式VR/AR体验非常重要，但是最近神经方法进展仍然缺乏在大动态场景下操纵视觉感知的编辑能力。为填补当前的......
  • 【C#编程技术总结】IO_Path类常用的API演示汇总
    目录​​ 一.IO_Path介绍​​​​二.Path解释​​​​三.常用API ​​​​四，其他Api​​  官方文档：​​https://learn.microsoft.com/zh-cn/dotnet/api/system.io.path?......
  • React+hook+ts+ant design封装一个table的组件
    前言我是歌谣我有个兄弟巅峰的时候排名c站总榜19叫前端小歌谣曾经我花了三年的时间创作了他现在我要用五年的时间超越他今天又是接近兄弟的一天人生难免坎坷大不了从......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99