Flink SQL -- 命令行的使用

1、启动Flink SQL
首先启动Flink的集群，选择独立集群模式或者是session的模式。此处选择是时session的模式：yarn-session.sh -d 在启动Flink SQL的client：
sql-client.sh
2、kafka SQL 连接器
在使用kafka作为数据源的时候需要上传jar包到flnik的lib下：/usr/local/soft/flink-1.15.2/lib可以去官网找对应的版本下载上传。

 

1、创建表：再流上定义表
再flink中创建表相当于创建一个视图（视图中不存数据，只有查询视图时才会去原表中读取数据）CREATE TABLE students (sid STRING,name STRING,age INT,sex STRING,clazz STRING    
) WITH ('connector' = 'kafka','topic' = 'student','properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092','properties.group.id' = 'testGroup','scan.startup.mode' = 'earliest-offset','format' = 'csv'
)2、查询数据(连续查询)：select clazz,count(1) as c from students group by clazz;
3、客户端为维护和可视化结果提供了三种的模式：

        1、表格模式（默认使用的模式），（table mode），在内存中实体化结果，并将结果用规则的分页表格可视化展示出来

SET 'sql-client.execution.result-mode' = 'table';

        2、变更日志模式，（changelog mode），不会实体化和可视化结果，而是由插入（+）和撤销（-）组成的持续查询产生结果流。

SET 'sql-client.execution.result-mode' = 'changelog';

        3、Tableau模式（tableau mode）更接近传统的数据库，会将执行的结果以制表的形式直接打在屏幕之上。具体显示的内容会取决于作业 执行模式的不同(execution.type)：

SET 'sql-client.execution.result-mode' = 'tableau';

4、 Flink SQL流批一体：
        1、流处理：

                a、流处理即可以处理有界流也可以处理无界流

                b、流处理的输出的结果是连续的结果

                c、流处理的底层是持续流的模型，上游的Task和下游的Task同时启动等待数据的到达

SET 'execution.runtime-mode' = 'streaming';
        2、批处理：

                a、批处理只能用于处理有界流

                b、输出的是最终的结果

                c、批处理的底层是MapReduce模型，会先执行上游的Task，在执行下游的Task 

SET 'execution.runtime-mode' = 'batch';
Flink做批处理，读取一个文件：-- 创建一个有界流的表
CREATE TABLE students_hdfs (sid STRING,name STRING,age INT,sex STRING,clazz STRING
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/spark/stu/students.txt',  -- 必选：指定路径'format' = 'csv'                     -- 必选：文件系统连接器指定 format
);select clazz,count(1) as c from 
students_hdfs
group by clazz
5、Flink SQL的连接器：
        1、kafka SQL 连接器

对于一些参数需要从官网进行了解。

                1、kafka source 

-- 创建kafka 表
CREATE TABLE students_kafka (`offset` BIGINT METADATA VIRTUAL, -- 偏移量`event_time` TIMESTAMP(3) METADATA FROM 'timestamp', --数据进入kafka的时间，可以当作事件时间使用sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'kafka','topic' = 'students', -- 数据的topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表'properties.group.id' = 'testGroup', -- 消费者组'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset'format' = 'csv' -- 读取数据的格式
);

                2、kafka sink 

-- 创建kafka 表
CREATE TABLE students_kafka_sink (sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'kafka','topic' = 'students_sink', -- 数据的topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表'properties.group.id' = 'testGroup', -- 消费者组'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset'format' = 'csv' -- 读取数据的格式
);-- 将查询结果保存到kafka中
insert into students_kafka_sink
select * from students_hdfs;kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic students_sink

        3、将更新的流写入到kafka中 

因为在Kafka是一个消息队列，是不会去重的。所以只需要将读取数据的格式改成canal-json。当数据被读取回来还是原来的流模式。

CREATE TABLE clazz_num_kafka (clazz STRING,num BIGINT
) WITH ('connector' = 'kafka','topic' = 'clazz_num', -- 数据的topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表'properties.group.id' = 'testGroup', -- 消费者组'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset'format' = 'canal-json' -- 读取数据的格式
);-- 将更新的数据写入kafka需要使用canal-json格式，数据中会带上操作类型
{"data":[{"clazz":"文科一班","num":71}],"type":"INSERT"}
{"data":[{"clazz":"理科三班","num":67}],"type":"DELETE"}insert into clazz_num_kafka
select clazz,count(1) as num from 
students
group by clazz;kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic clazz_num
        2、 hdfs SQL 连接器

                1、hdfs source

                        Flink读取文件可以使用有界流的方式，也可以是无界流方式。

-- 有界流
CREATE TABLE students_hdfs_batch (sid STRING,name STRING,age INT,sex STRING,clazz STRING
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径'format' = 'csv'                     -- 必选：文件系统连接器指定 format
);select * from students_hdfs_batch;-- 无界流
-- 基于hdfs做流处理，读取数据是以文件为单位，延迟比kafka大
CREATE TABLE students_hdfs_stream (sid STRING,name STRING,age INT,sex STRING,clazz STRING
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径'format' = 'csv' ,                    -- 必选：文件系统连接器指定 format'source.monitor-interval' = '5000' -- 每隔一段时间扫描目录，生成一个无界流
);select * from students_hdfs_stream;

                2、hdfs sink

-- 1、批处理模式(使用方式和底层原理和hive类似)
SET 'execution.runtime-mode' = 'batch';-- 创建表
CREATE TABLE clazz_num_hdfs (clazz STRING,num BIGINT
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/clazz_num',  -- 必选：指定路径'format' = 'csv'                     -- 必选：文件系统连接器指定 format
);
-- 将查询结果保存到表中
insert into clazz_num_hdfs
select clazz,count(1) as num
from students_hdfs_batch
group by clazz;-- 2、流处理模式
SET 'execution.runtime-mode' = 'streaming'; -- 创建表，如果查询数据返回的十更新更改的流需要使用canal-json格式
CREATE TABLE clazz_num_hdfs_canal_json (clazz STRING,num BIGINT
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/clazz_num_canal_json',  -- 必选：指定路径'format' = 'canal-json'                     -- 必选：文件系统连接器指定 format
);insert into clazz_num_hdfs_canal_json
select clazz,count(1) as num
from students_hdfs_stream
group by clazz;
3、MySQL SQL 连接器

        1、整合：

# 1、上传依赖包到flink 的lib目录下/usr/local/soft/flink-1.15.2/lib
flink-connector-jdbc-1.15.2.jar
mysql-connector-java-5.1.49.jar# 2、需要重启flink集群
yarn application -kill [appid]
yarn-session.sh -d# 3、重新进入sql命令行
sql-client.sh

         2、mysql   source 

-- 有界流
-- flink中表的字段类型和字段名需要和mysql保持一致
CREATE TABLE students_jdbc (id BIGINT,name STRING,age BIGINT,gender STRING,clazz STRING,PRIMARY KEY (id) NOT ENFORCED -- 主键
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://master:3306/student','table-name' = 'students','username' ='root','password' ='123456'
);select * from students_jdbc;

        3、mysql sink 

-- 创建kafka 表
CREATE TABLE students_kafka (`offset` BIGINT METADATA VIRTUAL, -- 偏移量`event_time` TIMESTAMP(3) METADATA FROM 'timestamp', --数据进入kafka的时间，可以当作事件时间使用sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'kafka','topic' = 'students', -- 数据的topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表'properties.group.id' = 'testGroup', -- 消费者组'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset'format' = 'csv' -- 读取数据的格式
);-- 创建mysql sink表
CREATE TABLE clazz_num_mysql (clazz STRING,num BIGINT,PRIMARY KEY (clazz) NOT ENFORCED -- 主键
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://master:3306/student','table-name' = 'clazz_num','username' ='root','password' ='123456'
);--- 再mysql创建接收表
CREATE TABLE clazz_num (clazz varchar(10),num BIGINT,PRIMARY KEY (clazz) -- 主键
) ;-- 将sql查询结果实时写入mysql
-- 将更新更改的流写入mysql,flink会自动按照主键更新数据
insert into clazz_num_mysql
select 
clazz,
count(1) as num from 
students_kafka
group by clazz;kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic students 插入一条数据
        4、DataGen：用于生成随机数据，一般用在高性能测试上
-- 创建包（只能用于source表）
CREATE TABLE students_datagen (sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'datagen','rows-per-second'='5', -- 每秒随机生成的数据量'fields.age.min'='1','fields.age.max'='100','fields.sid.length'='10','fields.name.length'='2','fields.sex.length'='1','fields.clazz.length'='4'
);

        5、print：用于高性能测试 只能用于sink表
CREATE TABLE print_table (sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'print'
);insert into print_table
select * from students_datagen;结果需要在提交的任务中查看。
        6、BlackHole ：是用于高性能测试使用，在后面可以用于Flink的反压的测试。
CREATE TABLE blackhole_table (sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'blackhole'
);insert into blackhole_table
select * from students_datagen;
6、SQL 语法
        1、Hints：

               用于提示执行，在Flink中可以动态的修改表中的属性，在Spark中可以用于广播。在修改动态表中属性后，不需要在重新建表，就可以读取修改后的需求。

CREATE TABLE students_kafka (`offset` BIGINT METADATA VIRTUAL, -- 偏移量`event_time` TIMESTAMP(3) METADATA FROM 'timestamp', --数据进入kafka的时间，可以当作事件时间使用sid STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'kafka','topic' = 'students', -- 数据的topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表'properties.group.id' = 'testGroup', -- 消费者组'scan.startup.mode' = 'latest-offset', -- 读取数据的位置earliest-offset latest-offset'format' = 'csv' -- 读取数据的格式
);-- 动态修改表属性，可以在查询数据时修改读取kafka数据的位置，不需要重新创建表
select * from students_kafka /*+ OPTIONS('scan.startup.mode' = 'earliest-offset') */;-- 有界流
CREATE TABLE students_hdfs (sid STRING,name STRING,age INT,sex STRING,clazz STRING
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径'format' = 'csv'                     -- 必选：文件系统连接器指定 format
);-- 可以在查询hdfs时，不需要再重新的创建表就可以动态改成无界流
select * from students_hdfs /*+ OPTIONS('source.monitor-interval' = '5000' )  */;
         2、WITH:

                当一段SQL语句在被多次使用的时候，就将通过with给这个SQL起一个别名，类似于封装起来，就是为这个SQL创建一个临时的视图（并不是真正的视图），方便下次使用。

CREATE TABLE students_hdfs (sid STRING,name STRING,age INT,sex STRING,clazz STRING
)WITH ('connector' = 'filesystem',           -- 必选：指定连接器类型'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径'format' = 'csv'                     -- 必选：文件系统连接器指定 format
);-- 可以在查询hdfs时，不需要再重新的创建表就可以动态改成无界流
select * from students_hdfs /*+ OPTIONS('source.monitor-interval' = '5000' )  */;-- tmp别名代表的时子查询的sql,可以在后面的sql中多次使用
with tmp as (select * from students_hdfs /*+ OPTIONS('source.monitor-interval' = '5000' )  */where clazz='文科一班'
)
select * from tmp
union all
select * from tmp;
        3、DISTINCT:

在flink 的流处理中，使用distinct，flink需要将之前的数据保存在状态中，如果数据一直增加，状态会越来越大 状态越来越大，checkpoint时间会增加，最终会导致flink任务出问题

select 
count(distinct sid) 
from students_kafka /*+ OPTIONS('scan.startup.mode' = 'earliest-offset') */;select count(sid)  
from (select distinct *from students_kafka /*+ OPTIONS('scan.startup.mode' = 'earliest-offset') */
);

注意事项：

       1、 当Flink Client客户端退出来以后，里面创建的动态表就不存在了。这些表结构是元数据，是存储在内存中的。

        2、当在进行where过滤的时候，字符串会出现三种情况：空的字符串、空格字符串、null的字符串，三者是有区别的：

        这三者是不同的概念，在进行where过滤的时候过滤的条件是不同的。

1、过滤空的字符串：where s！= ‘空字符串’2、过滤空格字符串：where s！= ‘空格’3、过滤null字符串：where s！= null
Flink SQL中常见的函数：from_unixtime： 以字符串格式 string 返回数字参数 numberic 的表示形式（默认为 ‘yyyy-MM-dd HH:mm:ss’to_timestamp：  将格式为 string2（默认为：‘yyyy-MM-dd HH:mm:ss’）的字符串 string1 转换为 timestamp

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/141958.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

LabVIEW中NIGPIB设备与驱动程序不相关的MAX报错

LabVIEW中NIGPIB设备与驱动程序不相关的MAX报错

LabVIEW中NIGPIB设备与驱动程序不相关的MAX报错 当插入GPIB-USB设备时，看到了NI MAX中列出该设备，但却显示了黄色警告指示，并且指出Windows没有与您的设备相关的驱动程序。 解决方案 需要安装能兼容的NI-488.2驱动程序。 通过交叉参考以下有…
阅读更多...
【C++初阶（八）】C/C++内存管理详解

【C++初阶（八）】C/C++内存管理详解

本专栏内容为：C学习专栏，分为初阶和进阶两部分。 通过本专栏的深入学习，你可以了解并掌握C。 💓博主csdn个人主页：小小unicorn ⏩专栏分类：C 🚚代码仓库：小小unicorn的代码仓库&…
阅读更多...
Leetcode刷题详解——不同路径 III

Leetcode刷题详解——不同路径 III

1. 题目链接：980. 不同路径 III 2. 题目描述： 在二维网格 grid 上，有 4 种类型的方格： 1 表示起始方格。且只有一个起始方格。2 表示结束方格，且只有一个结束方格。0 表示我们可以走过的空方格。-1 表示我们无法跨越的…
阅读更多...
线代笔记本

线代笔记本

阅读更多...
【读点论文】结构化剪枝

【读点论文】结构化剪枝

结构化剪枝 在一个神经网络模型中，通常包含卷积层、汇合层、全连接层、非线形层等基本结构，通过这些基本结构的堆叠，最终形成我们所常用的深度神经网络。 早在 1998 年，LeCun 等人使用少数几个基本结构组成 5 层的 LeNet-5 网络&…
阅读更多...
Python爬虫过程中DNS解析错误解决策略

Python爬虫过程中DNS解析错误解决策略

在Python爬虫开发中，经常会遇到DNS解析错误，这是一个常见且也令人头疼的问题。DNS解析错误可能会导致爬虫失败，但幸运的是，我们可以采取一些策略来处理这些错误，确保爬虫能够正常运行。本文将介绍什么是DNS解析错误&am…
阅读更多...
SpringBoot从零到一项目实战落地博客系统(附源码！！！)

SpringBoot从零到一项目实战落地博客系统(附源码！！！)

1.项目内容 1.1.页面展示 1.2.博客分类 1.3.面试辅导 1.4.私教带徒 1.5.文章编辑 1.6.后台管理 2.项目架构及技术描述 2.1.本项目用到的技术和框架 项目构建：Mavenweb框架：Springboot数据库ORM：Mybatis数据库连接池： HikariCP分…
阅读更多...
[Android]修改应用包名、名称、版本号、Icon以及环境判断和打包

[Android]修改应用包名、名称、版本号、Icon以及环境判断和打包

1.修改包名 在Android Studio中更改项目的包名涉及几个步骤： 打开项目结构: 在Android Studio中，确保您处于Android视图模式（在左侧面板顶部有一个下拉菜单可以选择）。 重命名包名: 在项目视图中，找到您的包名&…
阅读更多...
论文导读 | 融合大规模语言模型与知识图谱的推理方法

论文导读 | 融合大规模语言模型与知识图谱的推理方法

前 言 大规模语言模型在多种自然语言处理相关任务上展现了惊人的能力，如智能问答等，但是其推理能力尚未充分展现。本文首先介绍大模型进行推理的经典方法，然后进一步介绍知识图谱与大模型融合共同进行推理的工作。 文章一：使用思维…
阅读更多...
好消息！2023年汉字小达人市级比赛在线模拟题大更新：4个组卷+11个专项，助力孩子更便捷、有效、有趣地备赛

好消息！2023年汉字小达人市级比赛在线模拟题大更新：4个组卷+11个专项，助力孩子更便捷、有效、有趣地备赛

自从《中文自修》杂志社昨天发通知，官宣了2023年第十届汉字小达人市级比赛的日期和安排后，各路学霸们闻风而动，在自己本就繁忙的日程中又加了一项：备赛汉字小达人市级比赛，11月30日，16点-18点。 根据这几年…
阅读更多...
C 语言指针怎么理解？

C 语言指针怎么理解？

今日话题，C 语言指针怎么理解？让我用更简洁的方式来表达这个内容：就像桌面上的快捷方式一样，指针也可以有多层引用。我们可以将指针比作快捷方式的图标，快捷方式可以指向游戏（普通指针）&#xf…
阅读更多...
【JavaEE初阶】IP协议简介

【JavaEE初阶】IP协议简介

文章目录 前言🌴IP协议的概念🌳IP数据报🚩IPv4协议头格式🚩IPv6的诞生 🎍IP地址🚩IP地址的格式：🚩IP地址的分类🎈网络号与主机号的划分 🚩特殊的IP地址&#…
阅读更多...
【机器学习】八、规则学习

【机器学习】八、规则学习

知识图谱与基本概念 基本概念 规则学习定义：从训练数据中学习出一组能用于对未见示例进行判别的规则。 规则定义：规则一般是：语义明确、能描述数据分布所隐含的客观规律或领域概念。 逻辑规则定义：⊕←?1⋀?2⋀?3…⋀??⊕…
阅读更多...
任意注册漏洞

任意注册漏洞

目录 一漏洞介绍 二实战演示 三漏洞修复 本文由掌控安全学院 - 小博 投稿 一漏洞介绍 1.未验证邮箱/手机号 情景：应用为了方便用户记录用户名，使用邮箱和手机号作为用户名（因此很多应用在注册的时候就要求用户填写，多数时候…
阅读更多...
CTFSHOW -SQL 注入

CTFSHOW -SQL 注入

重新来做一遍 争取不看wp 还是看了。。。。 CTFshow sql注入 上篇(web171-220)更新中 - 掘金 【精选】CTFshow-WEB入门-SQL注入(上)_having盲注_bfengj的博客-CSDN博客 web171 基本联合注入 拿到题目我们已经知道了是sql注入 所以我们可以直接开始 第一题 不会难道哪里去…
阅读更多...
Pytorch常用的函数(四)深度学习中常见的上采样方法总结

Pytorch常用的函数(四)深度学习中常见的上采样方法总结

Pytorch常用的函数(四)深度学习中常见的上采样方法总结 我们知道在深度学习中下采样的方式比较常用的有两种： 池化 步长为2的卷积 而在上采样过程中常用的方式有三种： 插值 反池化 反卷积 不论是语义分割、目标检测还是三维重建等模型&#xff0…
阅读更多...
ios 对话框 弹框，输入对话框 普通对话框

ios 对话框 弹框，输入对话框 普通对话框

1 普通对话框 UIAlertController* alert [UIAlertController alertControllerWithTitle:"a" message:"alert12222fdsfs" pr…
阅读更多...
企业大楼门禁，千万不要这么管理！太慢了！

企业大楼门禁，千万不要这么管理！太慢了！

随着社会科技的飞速发展，安全管理已经成为各行业关注的焦点之一。在这个信息化时代，门禁监控系统作为一种全面提升安全性、管理效率的关键工具，逐渐成为企事业单位、学校、医疗机构等场所的不可或缺的一部分。 传统的门禁系统已经无法满足现代…
阅读更多...
【移远QuecPython】EC800M物联网开发板的硬件TIM定时器精准延时

【移远QuecPython】EC800M物联网开发板的硬件TIM定时器精准延时

【移远QuecPython】EC800M物联网开发板的硬件TIM定时器精准延时 文章目录 导入库定时器初始化延时函数定时中断回调调用函数打包附录：列表的赋值类型和py打包列表赋值BUG复现代码改进优化总结 py打包 首先 这个定时器是硬件底层级别的 优先级最高 如果调用 会导致GN…
阅读更多...
JavaScript库：jQuery，简化编程

JavaScript库：jQuery，简化编程

jQuery介绍 官方网站: https://jquery.com jQuery 是一个 JavaScript 库 。极大地简化了 JavaScript 编程，例如 JS 原生代码几十行 实现的功 能， jQuery 可能一两行就可以实现，因此得到前端程序猿广泛应用。（现在处在比较边…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023