ToB企服应用市场:ToB评测及商务社交产业平台

标题: 基于docker安装flink [打印本页]

---

作者: 天津储鑫盛钢材现货供应商    时间: 2024-8-19 12:59
标题: 基于docker安装flink
环境准备

Flink

docker-compose方式

1. version: "3"
2. services:
3.   jobmanager:
4.     image: flink:latest
5.     expose:
6.       - "6123"
7.     ports:
8.       - "8081:8081"
9.     command: jobmanager
10.     environment:
11.       - JOB_MANAGER_RPC_ADDRESS=jobmanager
13.   taskmanager:
14.     image: flink:latest
15.     expose:
16.       - "6121"
17.       - "6122"
18.     depends_on:
19.       - jobmanager
20.     command: taskmanager
21.     links:
22.       - "jobmanager:jobmanager"
23.     environment:
24.       - JOB_MANAGER_RPC_ADDRESS=jobmanager

复制代码

前端访问地点： http://192.168.56.112:8081/#/overview
二进制部署

1. wget https://archive.apache.org/dist/flink/flink-1.13.3/flink-1.13.3-bin-scala_2.11.tgz
3. vim conf/flink-conf.yaml
5. jobmanager.rpc.address: 192.168.56.112 # 修改为本机ip
7. ./bin/start-cluster.sh

复制代码

Kafka

1. version: '2'
2. services:
3.   zookeeper:
4.     image: wurstmeister/zookeeper   ## 镜像
5.     ports:
6.       - "2181:2181"                 ## 对外暴露的端口号
7.   kafka:
8.     image: wurstmeister/kafka       ## 镜像
9.     volumes:
10.         - /etc/localtime:/etc/localtime ## 挂载位置（kafka镜像和宿主机器之间时间保持一直）
11.     ports:
12.       - "9092:9092"
13.     environment:
14.       KAFKA_ADVERTISED_HOST_NAME: 192.168.56.112    ## 修改:宿主机IP
15.       KAFKA_ZOOKEEPER_CONNECT: 192.168.56.112:2181       ## 卡夫卡运行是基于zookeeper的
16.   kafka-manager:
17.     image: sheepkiller/kafka-manager                ## 镜像：开源的web管理kafka集群的界面
18.     environment:
19.         ZK_HOSTS:                    ## 修改:宿主机IP
20.     ports:
21.       - "9000:9000"

复制代码

Mysql

1. docker run -d -p3306:3306 --name=mysql57 -e MYSQL_ROOT_PASSWORD=111111 mysql:5.7

复制代码

Flink 实行 SQL下令

进入SQL客户端CLI

1. docker exec  -it flink_jobmanager_1  /bin/bash
3. ./bin/sql-client.sh

复制代码

实行SQL查询

1. SELECT 'Hello World';

复制代码

表格模式

表格模式（table mode）在内存中物化结果，并将结果用规则的分页表格的形式可视化展示出来。实行如下下令启用：

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

复制代码

可以使用如下查询语句检察差异模式的的运行结果：

1. SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name;

复制代码

变更日记模式

变更日记模式（changelog mode）不会物化结果。可视化展示由插入（+）和撤销（-）构成的持续查询结果流。

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

复制代码

Tableau模式

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

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

复制代码

   注意：当你在流式查询上使用这种模式时，Flink 会将结果持续的打印在当前的控制台上。如果流式查询的输入是有限数据集，那么 Flink 在处理惩罚完全部的输入数据之后，作业会自动停止，同时控制台上的打印也会自动停止。如果你想提前竣事这个查询，那么可以直接使用 CTRL-C 按键，这个会停止作业同时停止在控制台上的打印。
  窗口计算

TUMBLE(time_attr, interval) 界说一个滚动窗口。滚动窗口把行分配到有固定持续时间（ interval ）的不重叠的连续窗口。比如，5 分钟的滚动窗口以 5 分钟为隔断对行举行分组。滚动窗口可以界说在事件时间（批处理惩罚、流处理惩罚）或处理惩罚时间（流处理惩罚）上。
窗口计算

TUMBLE(time_attr, interval) 界说一个滚动窗口。滚动窗口把行分配到有固定持续时间（ interval ）的不重叠的连续窗口。比如，5 分钟的滚动窗口以 5 分钟为隔断对行举行分组。滚动窗口可以界说在事件时间（批处理惩罚、流处理惩罚）或处理惩罚时间（流处理惩罚）上。
滚动窗口demo

根据订单信息使用kafka作为数据源表，JDBC作为数据结果表统计用户在5秒内的订单数目，并根据窗口的订单id和窗口开启时间作为主键，将结果及时统计到JDBC中：

• 在MySQL的flink数据库下创建表order_count，创建语句如下：
  

1. CREATE TABLE `flink`.`order_count` (
2.         `user_id` VARCHAR(32) NOT NULL,
3.         `window_start` TIMESTAMP NOT NULL,
4.         `window_end` TIMESTAMP NULL,
5.         `total_num` BIGINT UNSIGNED NULL,
6.         PRIMARY KEY (`user_id`, `window_start`)
7. )        ENGINE = InnoDB
8.         DEFAULT CHARACTER SET = utf8mb4
9.         COLLATE = utf8mb4_general_ci;

复制代码

• 创建flink opensource sql作业，并提交运行作业
  

1. CREATE TABLE orders (
2.   order_id string,
3.   order_channel string,
4.   order_time timestamp(3),
5.   pay_amount double,
6.   real_pay double,
7.   pay_time string,
8.   user_id string,
9.   user_name string,
10.   area_id string,
11.   watermark for order_time as order_time - INTERVAL '3' SECOND
12. ) WITH (
13.   'connector' = 'kafka',
14.   'topic' = 'order_topic',
15.   'properties.bootstrap.servers' = '192.168.56.112:9092',
16.   'properties.group.id' = 'order_group',
17.   'scan.startup.mode' = 'latest-offset',
18.   'format' = 'json'
19. );
21. CREATE TABLE jdbcSink (
22.   user_id string,
23.   window_start timestamp(3),
24.   window_end timestamp(3),
25.   total_num BIGINT,
26.   primary key (user_id, window_start) not enforced
27. ) WITH (
28.   'connector' = 'jdbc',
29.   'url' = 'jdbc:mysql://192.168.56.112:3306/flink',
30.   'table-name' = 'order_count',
31.   'username' = 'root',
32.   'password' = '111111',
33.   'sink.buffer-flush.max-rows' = '1'
34. );
36. SELECT
37.     'WINDOW',-- window_start,window_end,
38.     group_key,record_num,create_time,
39.     SUM(record_num) OVER w AS sum_amount
40. FROM temp
41. WINDOW w AS (
42.   PARTITION BY group_key
43.   ORDER BY rowtime
44.   RANGE BETWEEN INTERVAL '10' SECOND PRECEDING AND CURRENT ROW)
45.   
46. select
47.     user_id,
48.     TUMBLE_START(order_time, INTERVAL '5' SECOND),
49.     TUMBLE_END(order_time, INTERVAL '5' SECOND),
50.     COUNT(*) from orders
51.     GROUP BY user_id, TUMBLE(order_time, INTERVAL '5' SECOND) having count(*) > 3;
52.    
53. SELECT
54.     'WINDOW',
55.     user_id,order_id,real_pay,order_time
56.     COUNT(*) OVER w AS sum_amount
57. FROM orders
58. WINDOW w AS (
59.   PARTITION BY user_id
60.   ORDER BY order_time
61.   RANGE BETWEEN INTERVAL '60' SECOND PRECEDING AND CURRENT ROW)
62.    
64. insert into jdbcSink select
65.     user_id,
66.     TUMBLE_START(order_time, INTERVAL '5' SECOND),
67.     TUMBLE_END(order_time, INTERVAL '5' SECOND),
68.     COUNT(*) from orders
69.     GROUP BY user_id, TUMBLE(order_time, INTERVAL '5' SECOND) having count(*) > 3;

复制代码

• Kafka 相关操作
  

1. bin/kafka-topics.sh --zookeeper 192.168.56.112:2181 --list
3. bin/kafka-topics.sh --zookeeper 192.168.56.112:2181 --create --replication-factor 1 --partitions 1 --topic order_topic
5. bin/kafka-console-producer.sh --broker-list 192.168.56.112:9092 --topic order_topic
7. bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.112:9092 --topic order_topic --from-beginning
9. bin/kafka-topics.sh --zookeeper 192.168.56.112:2181 --describe --topic order_topic
11. bin/kafka-topics.sh --zookeeper 192.168.56.112:2181 --delete --topic order_topic

复制代码

发送数据样例

1. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-09-26 15:20:11", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
2. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:28:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
3. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:29:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
4. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:29:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
5. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:29:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
6. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:30:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
7. {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2023-08-10 17:30:10", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}

复制代码

滑动窗口

1. SELECT * FROM TABLE(
2.     HOP(TABLE orders, DESCRIPTOR(order_time), INTERVAL '2' SECOND, INTERVAL '10' SECOND));
4. SELECT * FROM TABLE(
5.     HOP(
6.       DATA => TABLE orders,
7.       TIMECOL => DESCRIPTOR(order_time),
8.       SLIDE => INTERVAL '5' MINUTES,
9.       SIZE => INTERVAL '10' MINUTES));
10.       
11.       
12. SELECT window_start, window_end, SUM(pay_amount)
13.   FROM TABLE(
14.     HOP(TABLE orders, DESCRIPTOR(order_time), INTERVAL '2' SECOND, INTERVAL '10' SECOND))
15.   GROUP BY window_start, window_end;

复制代码

踩坑

• Could not find any factory for identifier ‘kafka’ that implements ‘org.apache.flink.table.factories.DynamicTableFactory’ in the classpath.
  

检察flink version
flink-sql-connector-kafka-1.17.1.jar
https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka/1.17.1
下载对应版本jar，放到lib目次下，重启

• Could not find any factory for identifier ‘jdbc’ that implements 'org.apache.flink.table.factories.DynamicTableFactory
  flink-connector-jdbc-3.1.0-1.17.jar
  https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc/3.1.0-1.17/flink-connector-jdbc-3.1.0-1.17.jar
  
• Caused by: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
  

https://mvnrepository.com/artifact/com.mysql/mysql-connector-j/8.0.31

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4