• Flink SQL -- 命令行的使用

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

     

    1. 1、创建表：
    2.  
    3. 再流上定义表
    4. 再flink中创建表相当于创建一个视图（视图中不存数据，只有查询视图时才会去原表中读取数据）
    5.  
    6.  
    7. CREATE TABLE students (
    8.     sid STRING,
    9.     name STRING,
    10.     age INT,
    11.     sex STRING,
    12.     clazz STRING    
    13. ) WITH (
    14.   'connector' = 'kafka',
    15.   'topic' = 'student',
    16.   'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
    17.   'properties.group.id' = 'testGroup',
    18.   'scan.startup.mode' = 'earliest-offset',
    19.   'format' = 'csv'
    20. )
    21.  
    22.  
    23. 2、查询数据(连续查询)：
    24.  
    25. select clazz,count(1) as c from students group by clazz;
    26.  
    27.
    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';
    1. Flink做批处理，读取一个文件：
    2.  
    3. -- 创建一个有界流的表
    4. CREATE TABLE students_hdfs (
    5.     sid STRING,
    6.     name STRING,
    7.     age INT,
    8.     sex STRING,
    9.     clazz STRING
    10. )WITH (
    11.   'connector' = 'filesystem',           -- 必选：指定连接器类型
    12.   'path' = 'hdfs://master:9000/data/spark/stu/students.txt',  -- 必选：指定路径
    13.   'format' = 'csv'                     -- 必选：文件系统连接器指定 format
    14. );
    15.  
    16.  
    17. select clazz,count(1) as c from 
    18. students_hdfs
    19. group by clazz
    5、Flink SQL的连接器：
            1、kafka SQL 连接器

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

                    1、kafka source 

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

                    2、kafka sink 

    1. -- 创建kafka 表
    2. CREATE TABLE students_kafka_sink (
    3.     sid STRING,
    4.     name STRING,
    5.     age INT,
    6.     sex STRING,
    7.     clazz STRING
    8. ) WITH (
    9.   'connector' = 'kafka',
    10.   'topic' = 'students_sink', -- 数据的topic
    11.   'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- broker 列表
    12.   'properties.group.id' = 'testGroup', -- 消费者组
    13.   'scan.startup.mode' = 'earliest-offset', -- 读取数据的位置earliest-offset latest-offset
    14.   'format' = 'csv' -- 读取数据的格式
    15. );
    16.  
    17. -- 将查询结果保存到kafka中
    18. insert into students_kafka_sink
    19. select * from students_hdfs;
    20.  
    21. kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic students_sink

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

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

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

                    1、hdfs source

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

    1. -- 有界流
    2. CREATE TABLE students_hdfs_batch (
    3.     sid STRING,
    4.     name STRING,
    5.     age INT,
    6.     sex STRING,
    7.     clazz STRING
    8. )WITH (
    9.   'connector' = 'filesystem',           -- 必选：指定连接器类型
    10.   'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径
    11.   'format' = 'csv'                     -- 必选：文件系统连接器指定 format
    12. );
    13.  
    14. select * from students_hdfs_batch;
    15.  
    16. -- 无界流
    17. -- 基于hdfs做流处理，读取数据是以文件为单位，延迟比kafka大
    18. CREATE TABLE students_hdfs_stream (
    19.     sid STRING,
    20.     name STRING,
    21.     age INT,
    22.     sex STRING,
    23.     clazz STRING
    24. )WITH (
    25.     'connector' = 'filesystem',           -- 必选：指定连接器类型
    26.     'path' = 'hdfs://master:9000/data/student',  -- 必选：指定路径
    27.     'format' = 'csv' ,                    -- 必选：文件系统连接器指定 format
    28.     'source.monitor-interval' = '5000' -- 每隔一段时间扫描目录，生成一个无界流
    29. );
    30.  
    31.  
    32. select * from students_hdfs_stream;

                    2、hdfs sink

    1. -- 1、批处理模式(使用方式和底层原理和hive类似)
    2. SET 'execution.runtime-mode' = 'batch';
    3.  
    4. -- 创建表
    5. CREATE TABLE clazz_num_hdfs (
    6.     clazz STRING,
    7.     num BIGINT
    8. )WITH (
    9.   'connector' = 'filesystem',           -- 必选：指定连接器类型
    10.   'path' = 'hdfs://master:9000/data/clazz_num',  -- 必选：指定路径
    11.   'format' = 'csv'                     -- 必选：文件系统连接器指定 format
    12. );
    13. -- 将查询结果保存到表中
    14. insert into clazz_num_hdfs
    15. select clazz,count(1) as num
    16. from students_hdfs_batch
    17. group by clazz;
    18.  
    19.  
    20. -- 2、流处理模式
    21. SET 'execution.runtime-mode' = 'streaming'; 
    22.  
    23. -- 创建表，如果查询数据返回的十更新更改的流需要使用canal-json格式
    24. CREATE TABLE clazz_num_hdfs_canal_json (
    25.     clazz STRING,
    26.     num BIGINT
    27. )WITH (
    28.   'connector' = 'filesystem',           -- 必选：指定连接器类型
    29.   'path' = 'hdfs://master:9000/data/clazz_num_canal_json',  -- 必选：指定路径
    30.   'format' = 'canal-json'                     -- 必选：文件系统连接器指定 format
    31. );
    32.  
    33. insert into clazz_num_hdfs_canal_json
    34. select clazz,count(1) as num
    35. from students_hdfs_stream
    36. group by clazz;
    3、MySQL SQL 连接器

            1、整合：

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

             2、mysql   source 

    1. -- 有界流
    2. -- flink中表的字段类型和字段名需要和mysql保持一致
    3. CREATE TABLE students_jdbc (
    4.     id BIGINT,
    5.     name STRING,
    6.     age BIGINT,
    7.     gender STRING,
    8.     clazz STRING,
    9.     PRIMARY KEY (id) NOT ENFORCED -- 主键
    10. ) WITH (
    11.     'connector' = 'jdbc',
    12.     'url' = 'jdbc:mysql://master:3306/student',
    13.     'table-name' = 'students',
    14.     'username' ='root',
    15.     'password' ='123456'
    16. );
    17.  
    18. select * from students_jdbc;

            3、mysql sink 

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

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

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

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

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

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

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

    1. select 
    2. count(distinct sid) 
    3. from students_kafka /*+ OPTIONS('scan.startup.mode' = 'earliest-offset') */;
    4.  
    5. select 
    6.     count(sid)  
    7. from (
    8.     select 
    9.     distinct *
    10.     from students_kafka /*+ OPTIONS('scan.startup.mode' = 'earliest-offset') */
    11. );

    注意事项：

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

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

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

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

  • 相关阅读:
    Google Earth Engine —— 1986-2020年植被覆盖度一元线性回归分析（黄河流域上游为例）
    电脑蓝屏怎么办 七大原因及解决办法来帮你
    图片的懒加载
    el-tree 获取过滤后的树结构
    GaussDB数据库SQL系列：DROP & TRUNCATE & DELETE
    入门数据库Days5
    const char *p,char const *p和char *const p区别
    如何在Android项目中制作和使用三方包(jar文件)
    IceRPC之调用管道与传出请求->快乐的RPC
    从原理到应用，人人都懂的ChatGPT指南
  • 原文地址：https://blog.csdn.net/m0_62078954/article/details/134321431