当前位置: 首页 > news >正文

Flink流批一体计算(13):PyFlink Tabel API之SQL DDL

1. TableEnvironment

创建 TableEnvironment

from pyflink.table import Environmentsettings, TableEnvironment# create a streaming TableEnvironmentenv_settings = Environmentsettings.in_streaming_mode()table_env = TableEnvironment.create(env_settings)# or create a batch TableEnvironmentenv_settings = Environmentsettings.in_batch_mode()table_env = TableEnvironment.create(env_settings)

TableEnvironment 是 Table API 和 SQL 集成的核心概念。

TableEnvironment 可以用来:

  • ·创建 Table
  • ·将 Table 注册成临时表
  • ·执行 SQL 查询
  • ·注册用户自定义的 (标量,表值,或者聚合) 函数
  • ·配置作业
  • ·管理 Python 依赖
  • ·提交作业执行

创建 source 表

table_env.execute_sql("""CREATE TABLE datagen (id INT,data STRING) WITH ('connector' = 'datagen','fields.id.kind' = 'sequence','fields.id.start' = '1','fields.id.end' = '10')""")

创建 sink 表

table_env.execute_sql("""CREATE TABLE print (id INT,data STRING) WITH ('connector' = 'print')""")

2. Table

Table 是 Python Table API 的核心组件。Table 是 Table API 作业中间结果的逻辑表示。

一个 Table 实例总是与一个特定的 TableEnvironment 相绑定。

不支持在同一个查询中合并来自不同 TableEnvironments 的表,例如 join 或者 union 它们。

通过列表类型的对象创建

你可以使用一个列表对象创建一张表:

from pyflink.table import Environmentsettings, TableEnvironment# 创建 批 TableEnvironmentenv_settings = Environmentsettings.in_batch_mode()table_env = TableEnvironment.create(env_settings)table = table_env.from_elements([(1, 'Hi'), (2, 'Hello')])table.to_pandas()==>print(table.to_pandas())table = table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['id', 'data'])print(table.to_pandas())

通过 DDL 创建

你可以通过 DDL 创建一张表,execute_sql(stmt) 执行指定的语句并返回执行结果。

执行语句可以是 DDL/DML/DQL/SHOW/DESCRIBE/EXPLAIN/USE。

注意,对于 "INSERT INTO" 语句,这是一个异步操作,通常在向远程集群提交作业时才需要使用。

但是,如果在本地集群或者 IDE 中执行作业时,你需要等待作业执行完成。

from pyflink.table import Environmentsettings, TableEnvironment# 创建流 TableEnvironmentenv_settings = Environmentsettings.in_streaming_mode()table_env = TableEnvironment.create(env_settings)table_env.execute_sql("""CREATE TABLE random_source (id BIGINT,data TINYINT) WITH ('connector' = 'datagen','fields.id.kind'='sequence','fields.id.start'='1','fields.id.end'='3','fields.data.kind'='sequence','fields.data.start'='4','fields.data.end'='6')""")table = table_env.from_path("random_source")table.to_pandas()

通过 Catalog 创建

Catalog

Catalog提供了元数据信息,例如数据库、表、分区、视图以及数据库或其他外部系统中存储的函数和信息。

数据处理最关键的方面之一是管理元数据。

元数据可以是临时的,例如临时表、或者通过 TableEnvironment 注册的 UDF。

元数据也可以是持久化的,例如 Hive Metastore 中的元数据。

Catalog 提供了一个统一的API,用于管理元数据,并使其可以从 Table API 和 SQL 查询语句中来访问。

Catalog类型

GenericInMemoryCatalog

基于内存实现的 Catalog,所有元数据只在 session 的生命周期内可用。

JdbcCatalog

JdbcCatalog使得用户可以将 Flink 通过 JDBC 协议连接到关系数据库。

PostgresCatalog 是当前实现的唯一一种 JDBC Catalog。

HiveCatalog

HiveCatalog 有两个用途:作为原生 Flink 元数据的持久化存储,以及作为读写现有 Hive 元数据的接口。

警告 Hive Metastore 以小写形式存储所有元数据对象名称,GenericInMemoryCatalog 区分大小写。

用户自定义 Catalog

Catalog 是可扩展的,用户可以通过实现 Catalog 接口来开发自定义 Catalog。

想要在 SQL CLI 中使用自定义 Catalog,用户除了需要实现自定义的 Catalog 之外,还需要为这个 Catalog 实现对应的 CatalogFactory 接口。

CatalogFactory 定义了一组属性,用于 SQL CLI 启动时配置 Catalog。

这组属性集将传递给发现服务,在该服务中,服务会尝试将属性关联到 CatalogFactory 并初始化相应的 Catalog 实例。

创建 Flink 表并将其注册到 Catalog

使用 SQL DDL

用户可以使用 DDL 通过 Table API 或者 SQL Client 在 Catalog 中创建表。

from pyflink.table.catalog import HiveCatalog# Create a HiveCatalogcatalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")# Register the catalogt_env.register_catalog("myhive", catalog)# Create a catalog databaset_env.execute_sql("CREATE DATABASE mydb WITH (...)")# Create a catalog tablet_env.execute_sql("CREATE TABLE mytable (name STRING, age INT) WITH (...)")# should return the tables in current catalog and database.t_env.list_tables()

通过 SQL DDL 创建的表和视图, 例如 “create table …” 和 “create view …",都存储在 catalog 中。

你可以通过 SQL 直接访问 catalog 中的表。

使用 Java/Scala

用户可以用编程的方式使用Java 或者 Scala 来创建 Catalog 表。

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable
from pyflink.table.descriptors import Kafka
settings = Environmentsettings.in_batch_mode()
t_env = TableEnvironment.create(settings)
# Create a HiveCatalog
catalog = HiveCatalog("myhive", None, "<path_of_hive_conf>")
# Register the catalog
t_env.register_catalog("myhive", catalog)
# Create a catalog database
database = CatalogDatabase.create_instance({"k1": "v1"}, None)
catalog.create_database("mydb", database)
# Create a catalog table
schema = Schema.new_builder() \.column("name", DataTypes.STRING()) \.column("age", DataTypes.INT()) \.build()  
catalog_table = t_env.create_table("myhive.mydb.mytable", TableDescriptor.for_connector("kafka").schema(schema)// ….build())
# tables should contain "mytable"
tables = catalog.list_tables("mydb")

TableEnvironment 维护了一个使用标识符创建的表的 catalogs 映射。

Catalog 中的表既可以是临时的,并与单个 Flink 会话生命周期相关联,也可以是永久的,跨多个 Flink 会话可见。

如果你要用 Table API 来使用 catalog 中的表,可以使用 “from_path” 方法来创建 Table API 对象:

from_path(path)   通过指定路径下已注册的表来创建一个表,例如通过 create_temporary_view 注册表。

相关文章:

Flink流批一体计算(13):PyFlink Tabel API之SQL DDL

1. TableEnvironment 创建 TableEnvironment from pyflink.table import Environmentsettings, TableEnvironment# create a streaming TableEnvironmentenv_settings Environmentsettings.in_streaming_mode()table_env TableEnvironment.create(env_settings)# or create…...

java笔试手写算法面试题大全含答案

1.统计一篇英文文章单词个数。 public class WordCounting { public static void main(String[] args) { try(FileReader fr new FileReader("a.txt")) { int counter 0; boolean state false; int currentChar; while((currentChar fr.read()) ! -1) { i…...

点云平面拟合和球面拟合

一、介绍 In this tutorial we learn how to use a RandomSampleConsensus with a plane model to obtain the cloud fitting to this model. 二、代码 #include <iostream> #include <thread> #include <pcl/point_types.h> #include <pcl/common/io.…...

部署问题集合(十九)linux设置Tomcat、Docker,以及使用脚本开机自启(亲测)

前言 因为不想每次启动虚拟机都要手动启动一遍这些东西&#xff0c;所以想要设置成开机自启的状态 设置Tomcat开机自启 创建service文件 vi /etc/systemd/system/tomcat.service添加如下内容&#xff0c;注意修改启动脚本和关闭脚本的地址 [Unit] DescriptionTomcat9068 A…...

视觉SLAM:一直在入门,如何能精通,CV领域的绝境长城,

目录 前言 福利&#xff1a;文末有chat-gpt纯分享&#xff0c;无魔法&#xff0c;无限制 1 什么是SLAM&#xff1f; 2 为什么用SLAM&#xff1f; 3 视觉SLAM怎么实现&#xff1f; 4 前端视觉里程计 5 后端优化 6 回环检测 7 地图构建 8 结语 前言 上周的组会上&…...

【报错】yarn --version Unrecognized option: --version Error...

文章目录 问题分析解决问题 在使用 npm install -g yarn 全局安装 yarn 后,查看yarn 的版本号,报错如下 PS D:\global-data-display> yarn --version Unrecognized option: --version Error: Could...

二叉搜索树的(查找、插入、删除)

一、二叉搜索树的概念 二叉搜索树又称二叉排序树&#xff0c;它或者是一棵空树&#xff0c;或者是具有以下性质的二叉树: 1、若它的左子树不为空&#xff0c;则左子树上所有节点的值都小于根节点的值&#xff1b; 2、若它的右子树不为空&#xff0c;则右子树上所有节点的值都…...

电力虚拟仿真 | 高压电气试验VR教学系统

在科技进步的推动下&#xff0c;我们的教育方式也在发生着翻天覆地的变化。其中&#xff0c;虚拟现实&#xff08;VR&#xff09;技术的出现&#xff0c;为我们提供了一种全新的、富有沉浸感的学习和培训方式。特别是在电力行业领域&#xff0c;例如&#xff0c;电力系统的维护…...

innovus如何设置size only

我正在「拾陆楼」和朋友们讨论有趣的话题&#xff0c;你⼀起来吧&#xff1f; 拾陆楼知识星球入口 给instance设置size only属性命令如下: dbset [dbGet top.inst.name aa/bb -p] .dontTouch sizeOk 给一个module设置size only需要foreach循环一下: foreach inst [dbGet top.…...

Java之继承详解二

3.7 方法重写 3.7.1 概念 方法重写 &#xff1a;子类中出现与父类一模一样的方法时&#xff08;返回值类型&#xff0c;方法名和参数列表都相同&#xff09;&#xff0c;会出现覆盖效果&#xff0c;也称为重写或者复写。声明不变&#xff0c;重新实现。 3.7.2 使用场景与案例…...

国内常见的几款可视化Web组态软件

组态软件是一种用于控制和监控各种设备的软件&#xff0c;也是指在自动控制系统监控层一级的软件平台和开发环境。这类软件实际上也是一种通过灵活的组态方式&#xff0c;为用户提供快速构建工业自动控制系统监控功能的、通用层次的软件工具。通常用于工业控制&#xff0c;自动…...

通过 git上传到 gitee 仓库

介绍 Git是目前世界上最先进的分布式版本控制系统&#xff0c;有这么几个特点&#xff1a; 分布式 &#xff1a;是用来保存工程源代码历史状态的命令行工具。保存点 &#xff1a;保存点可以追溯源码中的文件&#xff0c;并能得到某个时间点上的整个工程项目额状态&#xff1b;…...

设置Windows主机的浏览器为wls2的默认浏览器

1. 准备工作 wsl是可以使用Windows主机上安装的exe程序&#xff0c;出于安全考虑&#xff0c;默认情况下改功能是无法使用。要使用的话&#xff0c;终端需要以管理员权限启动。 我这里以Windows Terminal为例&#xff0c;介绍如何默认使用管理员权限打开终端&#xff0c;具体…...

森林生物量(蓄积量)估算全流程

python森林生物量&#xff08;蓄积量&#xff09;估算全流程 一.哨兵2号获取/去云处理/提取参数1.1 影像处理与下载1.2 导入2A级产品1.3导入我们在第1步生成的云掩膜文件1.4.SNAP掩膜操作1.5采用gdal计算各类植被指数1.6 纹理特征参数提取 二.哨兵1号获取/处理/提取数据2.1 纹理…...

MySQL数据库概述

MySQL数据库概述 1 SQL SQL语句大小写不敏感。 SQL语句末尾应该使用分号结束。 1.1 SQL语句及相关操作示例 DDL&#xff1a;数据定义语言&#xff0c;负责数据库定义、数据库对象定义&#xff0c;由CREATE、ALTER与DROP三个语法所组成DML&#xff1a;数据操作语言&#xff…...

2023年国赛数学建模思路 - 案例:退火算法

文章目录 1 退火算法原理1.1 物理背景1.2 背后的数学模型 2 退火算法实现2.1 算法流程2.2算法实现 建模资料 ## 0 赛题思路 &#xff08;赛题出来以后第一时间在CSDN分享&#xff09; https://blog.csdn.net/dc_sinor?typeblog 1 退火算法原理 1.1 物理背景 在热力学上&a…...

怎么借助ChatGPT处理数据结构的问题

目录 使用ChatGPT进行数据格式化转换 代码示例 ChatGPT格式化数据提示语 代码示例 批量格式化数据提示语 代码示例 ChatGPT生成的格式化批处理代码 使用ChatGPT合并不同数据源的数据 合并数据提示语 自动合并数据提示语 ChatGPT生成的自动合并代码 结论 数据合并是…...

Docker容器无法启动 Cannot find /usr/local/tomcat/bin/setclasspath.sh

报错信息如下 解决办法 权限不够 加上--privileged 获取最大权限 docker run --privileged --name lenglianerqi -p 9266:8080 -v /opt/docker/lenglianerqi/webapps:/usr/local/tomcat/webapps/ -v /opt/docker/lenglianerqi/webapps/userfile:/usr/local/tomcat/webapps/u…...

Pytorch-day08-模型进阶训练技巧-checkpoint

PyTorch 模型进阶训练技巧 自定义损失函数动态调整学习率 典型案例&#xff1a;loss上下震荡 [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BndMyRX0-1692613806232)(attachment:image-2.png)] 1、自定义损失函数 1、PyTorch已经提供了很多常用…...

【ArcGIS Pro二次开发】(61):样式(Style)和符号(Symbol)

在 ArcGIS Pro SDK 中&#xff0c;地图要素符号&#xff08;Symbol&#xff09;和符号样式&#xff08;Style&#xff09;是2个很重要的概念。 【Symbol】是用于表示地图上不同类型的要素&#xff08;如点、线、面&#xff09;的图形化表示。 在地图中&#xff0c;各种要素都…...

FFmpeg 低延迟同屏方案

引言 在实时互动需求激增的当下&#xff0c;无论是在线教育中的师生同屏演示、远程办公的屏幕共享协作&#xff0c;还是游戏直播的画面实时传输&#xff0c;低延迟同屏已成为保障用户体验的核心指标。FFmpeg 作为一款功能强大的多媒体框架&#xff0c;凭借其灵活的编解码、数据…...

基于Java Swing的电子通讯录设计与实现:附系统托盘功能代码详解

JAVASQL电子通讯录带系统托盘 一、系统概述 本电子通讯录系统采用Java Swing开发桌面应用&#xff0c;结合SQLite数据库实现联系人管理功能&#xff0c;并集成系统托盘功能提升用户体验。系统支持联系人的增删改查、分组管理、搜索过滤等功能&#xff0c;同时可以最小化到系统…...

C++.OpenGL (14/64)多光源(Multiple Lights)

多光源(Multiple Lights) 多光源渲染技术概览 #mermaid-svg-3L5e5gGn76TNh7Lq {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-3L5e5gGn76TNh7Lq .error-icon{fill:#552222;}#mermaid-svg-3L5e5gGn76TNh7Lq .erro…...

Mysql中select查询语句的执行过程

目录 1、介绍 1.1、组件介绍 1.2、Sql执行顺序 2、执行流程 2.1. 连接与认证 2.2. 查询缓存 2.3. 语法解析&#xff08;Parser&#xff09; 2.4、执行sql 1. 预处理&#xff08;Preprocessor&#xff09; 2. 查询优化器&#xff08;Optimizer&#xff09; 3. 执行器…...

排序算法总结(C++)

目录 一、稳定性二、排序算法选择、冒泡、插入排序归并排序随机快速排序堆排序基数排序计数排序 三、总结 一、稳定性 排序算法的稳定性是指&#xff1a;同样大小的样本 **&#xff08;同样大小的数据&#xff09;**在排序之后不会改变原始的相对次序。 稳定性对基础类型对象…...

基于Java+VUE+MariaDB实现(Web)仿小米商城

仿小米商城 环境安装 nodejs maven JDK11 运行 mvn clean install -DskipTestscd adminmvn spring-boot:runcd ../webmvn spring-boot:runcd ../xiaomi-store-admin-vuenpm installnpm run servecd ../xiaomi-store-vuenpm installnpm run serve 注意&#xff1a;运行前…...

PostgreSQL——环境搭建

一、Linux # 安装 PostgreSQL 15 仓库 sudo dnf install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-$(rpm -E %{rhel})-x86_64/pgdg-redhat-repo-latest.noarch.rpm# 安装之前先确认是否已经存在PostgreSQL rpm -qa | grep postgres# 如果存在&#xff0…...

从面试角度回答Android中ContentProvider启动原理

Android中ContentProvider原理的面试角度解析&#xff0c;分为​​已启动​​和​​未启动​​两种场景&#xff1a; 一、ContentProvider已启动的情况 1. ​​核心流程​​ ​​触发条件​​&#xff1a;当其他组件&#xff08;如Activity、Service&#xff09;通过ContentR…...

离线语音识别方案分析

随着人工智能技术的不断发展&#xff0c;语音识别技术也得到了广泛的应用&#xff0c;从智能家居到车载系统&#xff0c;语音识别正在改变我们与设备的交互方式。尤其是离线语音识别&#xff0c;由于其在没有网络连接的情况下仍然能提供稳定、准确的语音处理能力&#xff0c;广…...

聚六亚甲基单胍盐酸盐市场深度解析:现状、挑战与机遇

根据 QYResearch 发布的市场报告显示&#xff0c;全球市场规模预计在 2031 年达到 9848 万美元&#xff0c;2025 - 2031 年期间年复合增长率&#xff08;CAGR&#xff09;为 3.7%。在竞争格局上&#xff0c;市场集中度较高&#xff0c;2024 年全球前十强厂商占据约 74.0% 的市场…...