【参赛作品95】DLI Flink SQL+kafka+（opengauss和mysql）进行电商实时业务数据分析

2024-03-12 19:20
文章标签 sql 进行 mysql 业务 作品 实时 参赛 flink kafka 数据分析 电商 opengauss database 95 dli

本文主要是介绍【参赛作品95】DLI Flink SQL+kafka+（opengauss和mysql）进行电商实时业务数据分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

作者：吴毅

电商实时业务数据分析案例，
测试环境：操作系统：EulerOS_2.8_ARM-20200308_20200308192159

硬件配置：4vCPUs | 16 GiB | kc1.xlarge.4 IP地址：192.168.0.2

在上一篇已经部署好docker和docker版的opengauss。

1 部署docker版的kafka，要找支持arm64的docker版kafka。

1.1 先部署支持arm64的zookeeper，命令如下：

 
docker pull pi4k8s/k8szk:v3.4.14
docker run -itd --name zookeeper -p 21810:2181 -e ZK_REPLICAS=1 -e ZK_HEAP_SIZE=512M --hostname=zk-1 -v /opt/moudles/zookeeper/data:/var/lib/zookeeper -v /opt/moudles/zookeeper/log:/var/log/zookeeper pi4k8s/k8szk:v3.4.14 /bin/bash -c "/usr/bin/zkGenConfig.sh && /usr/bin/zkServer.sh start-foreground"

1.2 部署支持arm64的kafka

 
docker pull iecedge/cp-kafka-arm64:5.0.1
docker run -d --name=kafka -p 1099:1099 -p 9092:9092 -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=192.168.0.2:21810 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.0.2:9092 -e KAFKA_JMX_PORT=1099 -e KAFKA_JMX_HOSTNAME=192.168.0.2 -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -e KAFKA_HEAP_OPTS="-Xmx512M -Xms512M" -v /opt/moudles/kafka/data:/var/lib/kafka/data iecedge/cp-kafka-arm64:5.0.1

image.png
** 1.3 部署完后，下载kafka的源码包：**

 
cd /opt/softwares
wget  https://archive.apache.org/dist/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar zxf /opt/softwares/kafka_2.12-2.3.0.tgz -C /opt/modules/
cd /opt/modules/kafka_2.12-2.3.0/bin

1.4 创建topic

 
./kafka-topics.sh --create --zookeeper 192.168.0.2:21810 --replication-factor 1 --partitions 1 --topic trade_order

1.5 查看创建topic的数目

 
./kafka-topics.sh --list --zookeeper 192.168.0.2:21810

1.6 kafka生产者

 
./kafka-console-producer.sh --broker-list 192.168.0.2:9092 --topic trade_order

1.7 kafka消费者

 
./kafka-console-consumer.sh --bootstrap-server 192.168.0.2:9092 --topic trade_order --from-beginning

特别注意要放开9092（Kafka连接端口）。
2. 连接opengauss和创建表
2.1 连接opengauss

 
gsql -d testdb -U wuyi -p 5432 –r

2.2 创建表trade_channel_collect

 
CREATE TABLE trade_channel_collect (begin_time VARCHAR(32) NOT NULL,channel_code VARCHAR(32) NOT NULL,channel_name VARCHAR(32) NULL,cur_gmv BINARY_DOUBLE NULL,cur_order_user_count BIGINT  NULL,cur_order_count BIGINT  NULL,last_pay_time VARCHAR(32) NULL,flink_current_time VARCHAR(32) NULL,PRIMARY KEY (begin_time, channel_code)
);
comment on table trade_channel_collect is'各渠道的销售总额实时统计';

image.png
3 创建 DLI 增强型跨源
3.1 在控制台单击“服务列表”，选择“数据湖探索”，单击进入DLI服务页面。
单击“队列管理”，在队列列表中您所创建的通用队列
image.png
3.2 单击“跨源连接” >“增强型跨源” >“创建”。
配置如下：
– 绑定队列：选择您所创建的通用队列。
– 虚拟私有云：选择 Kafka 与 opengauss 实例所在的VPC
– 子网：选择 Kafka 与 opengauss 实例所在的子网。
image.png
3.3 测试队列与kafka和opengauss实例连通性。
单击“队列管理”，选择您所使用的队列，单击“操作”列中的“更多” >“测试地址连通性”
image.png
image.png
3.4 创建并提交 Flink 作业
单击DLI控制台左侧“作业管理”，选择“Flink作业”。单击“创建作业”。
– 类型：选择作业类型为： Flink OpenSource SQL。
– 名称：自定义
image.png
代码：

 
--********************************************************************--
-- 数据源：trade_order_detail_info (订单详情宽表)
--********************************************************************--
create table trade_order_detail (order_id string,      -- 订单IDorder_channel string,   -- 渠道order_time string,     -- 订单创建时间pay_amount double,     -- 订单金额real_pay double,      -- 实际付费金额pay_time string,      -- 付费时间user_id string,      -- 用户IDuser_name string,     -- 用户名area_id string       -- 地区ID
) with ("connector.type" = "kafka","connector.version" = "0.10","connector.properties.bootstrap.servers" = "192.168.0.2:9092", -- Kafka连接地址-- "connector.properties.group.id" = "kafka-test",   -- Kafka groupID"connector.topic" = "test",     -- Kafka topic"format.type" = "json","connector.startup-mode" = "latest-offset"
);

––
– 结果表：trade_channel_collect (各渠道的销售总额实时统计)
––
create table trade_channel_collect(
begin_time date, –统计数据的开始时间
channel_code varchar, – 渠道编号
channel_name varchar, – 渠道名
–cur_gmv INTEGER, – 当天GMV
cur_order_user_count bigint, – 当天付款人数
cur_order_count bigint, – 当天付款订单数
last_pay_time varchar, – 最近结算时间
flink_current_time varchar,
primary key (begin_time, channel_code) not enforced
) with (
“connector.type” = “jdbc”,
“connector.url” = “jdbc:postgresql://192.168.0.2:15432/testdb”, – opengauss连接地址，jdbc格式
“connector.table” = “trade_channel_collect”, – opengauss表名
“connector.driver” = “org.postgresql.Driver”,
“connector.username” = “wuyi”, – opengauss用户名
“connector.password” = “xxxxx”, – opengauss密码
“connector.write.flush.max-rows” = “1000”,
“connector.write.flush.interval” = “1s”
);

––
– 临时中间表
––
create view tmp_order_detail
as
select *
, case when t.order_channel not in (“webShop”, “appShop”, “miniAppShop”) then “other”
else t.order_channel end as channel_code –重新定义统计渠道 只有四个枚举值[webShop、appShop、miniAppShop、other]
, case when t.order_channel = “webShop” then _UTF8"网页商城"
when t.order_channel = “appShop” then _UTF8"app商城"
when t.order_channel = “miniAppShop” then _UTF8"小程序商城"
else _UTF8"其他" end as channel_name –渠道名称
from (
select *
, row_number() over(partition by order_id order by order_time desc ) as rn –去除重复订单数据
, concat(substr(“2021-03-25 12:03:00”, 1, 10), " 00:00:00") as begin_time
, concat(substr(“2021-03-25 12:03:00”, 1, 10), " 23:59:59") as end_time
from trade_order_detail
where pay_time >= concat(substr(“2021-03-25 12:03:00”, 1, 10), " 00:00:00") –取今天数据，为了方便运行，这里使用"2021-03-25 12:03:00"替代cast(LOCALTIMESTAMP as string)
and real_pay is not null
) t
where t.rn = 1;

– 按渠道统计各个指标
insert into trade_channel_collect
select
‘begin_time’ –统计数据的开始时间
, cast(channel_code as varchar) as channel_code
, cast(channel_name as varchar) as channel_name
, cast((COALESCE(sum(real_pay), 0) as integer) as cur_gmv) –当天GMV
, ((count(distinct user_id) as bigint) as cur_order_user_count –当天付款人数
, ((count(1) as bigint) as cur_order_count) –当天付款订单数
, cast((max(pay_time) as last_pay_time) as varchar) –最近结算时间
, cast((LOCALTIMESTAMP as flink_current_time) as varchar) –flink任务中的当前时间
from tmp_order_detail
where pay_time >= concat(substr(“2021-03-25 12:03:00”, 1, 10), " 00:00:00")
group by begin_time, channel_code, channel_name;

image.png
只要是kafka出来的数据是string类型，而opengauss不支持string。使用cast()函数也无法转换成功。下面使用mysql数据库来连接一下。Mysql数据类型兼容性好。
4. docker版mysql的部署
4.1 安装mysql

 
docker search biarms/mysql:5.7.30
docker pull biarms/mysql:5.7.30
docker run -d --name mysql57 --hostname= mysql57  -e MYSQL_ROOT_PASSWORD=123456  -p 3306:3306 biarms/mysql:5.7.30

特别要注意，mysql5.7要mysql:5.7.30以上的版本才能支持arm64

 
docker exec -it mysql57 /bin/bash
docker cp mysql57:/etc/mysql/my.cnf  /root/
/etc/mysql/mysql.conf.d/mysqld.cnf
docker cp mysql57:/etc/mysql/mysql.conf.d/mysqld.cnf  /root/

 
vi /root/mysqld.cnf
[mysqld]
character_set_server=utf8
[client]
default-character-set=utf8
docker cp /root/mysql.cnf ,mysql:/etc/mysql/mysql.conf.d/
docker restart mysql57

 
reate database dli_demo DEFAULT CHARACTER SET = utf8mb4;
CREATE TABLE `dli_demo`.`trade_channel_collect` (`begin_time` VARCHAR(32) NOT NULL,`channel_code` VARCHAR(32) NOT NULL,`channel_name` VARCHAR(32) NULL,`cur_gmv` DOUBLE UNSIGNED NULL,`cur_order_user_count` BIGINT UNSIGNED NULL,`cur_order_count` BIGINT UNSIGNED NULL,`last_pay_time` VARCHAR(32) NULL,`flink_current_time` VARCHAR(32) NULL,PRIMARY KEY (`begin_time`, `channel_code`)
)   ENGINE = InnoDBDEFAULT CHARACTER SET = utf8mb4COLLATE = utf8mb4_general_ciCOMMENT = '各渠道的销售总额实时统计';

4.2 查看创建topic的数目

 
cd /opt/modules/kafka_2.12-2.3.0/bin
./kafka-topics.sh --list --zookeeper 192.168.0.2:21810
# # kafka生产者
./kafka-console-producer.sh --broker-list 192.168.0.2:9092 --topic trade_order
kafka消费者
./kafka-console-consumer.sh --bootstrap-server 192.168.0.2:9092 --topic trade_order --from-beginning

image.png
获得的数据连接华为云DLV，可以大屏数据可视化
image.png
通过对opengauss和mysql的操作，可以很容易发现，mysql对数据类型兼容比较好。而且在创建表时，在数据库，表名和字段加上单引号或双引号都能成功建表，但是opengauss缺不行。Opengauss虽然有很多特性，但是在string的转换不是很好用，找了cast()函数还有其他方法都无法转换，也有可能是我水平不行，原本还想写关于opengauss的AI方面的特性，但是由于这个问题花了一段时间。只能等下次活动再写一下关于AI的特性。

这篇关于【参赛作品95】DLI Flink SQL+kafka+（opengauss和mysql）进行电商实时业务数据分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/802309。

相关文章

SQL中的外键约束

SQL中的外键约束

外键约束用于表示两张表中的指标连接关系。外键约束的作用主要有以下三点： 1.确保子表中的某个字段（外键）只能引用父表中的有效记录2.主表中的列被删除时，子表中的关联列也会被删除3.主表中的列更新时，子表中的关联元素也会被更新 子表中的元素指向主表 以下是一个外键约束的实例展示
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
如何去写一手好SQL

如何去写一手好SQL

MySQL性能 最大数据量 抛开数据量和并发数，谈性能都是耍流氓。MySQL没有限制单表最大记录数，它取决于操作系统对文件大小的限制。 《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB，才推荐分库分表。性能由综合因素决定，抛开业务复杂度，影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考，并非铁律。 博主曾经操作过超过4亿行数据
阅读更多...
性能分析之MySQL索引实战案例

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题，本文将一起从代码层去分析为什么没有建立索引？ 开源ERP项目地址：https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置
阅读更多...
MySQL数据库宕机，启动不起来，教你一招搞定！

MySQL数据库宕机，启动不起来，教你一招搞定！

作者介绍：老苏，10余年DBA工作运维经验，擅长Oracle、MySQL、PG、Mongodb数据库运维（如安装迁移，性能优化、故障应急处理等）公众号：老苏畅谈运维欢迎关注本人公众号，更多精彩与您分享。 MySQL数据库宕机，数据页损坏问题，启动不起来，该如何排查和解决，本文将为你说明具体的排查过程。 查看MySQL error日志 查看 MySQL error日志，排查哪个表（表空间
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...
C#实战|大乐透选号器[6]：实现实时显示已选择的红蓝球数量

C#实战|大乐透选号器[6]：实现实时显示已选择的红蓝球数量

哈喽，你好啊，我是雷工。 关于大乐透选号器在前面已经记录了5篇笔记，这是第6篇； 接下来实现实时显示当前选中红球数量，蓝球数量； 以下为练习笔记。 01 效果演示 当选择和取消选择红球或蓝球时，在对应的位置显示实时已选择的红球、蓝球的数量； 02 标签名称 分别设置Label标签名称为：lblRedCount、lblBlueCount
阅读更多...
MySQL高性能优化规范

MySQL高性能优化规范

前言：      笔者最近上班途中突然想丰富下自己的数据库优化技能。于是在查阅了多篇文章后，总结出了这篇！ 数据库命令规范 所有数据库对象名称必须使用小写字母并用下划线分割 所有数据库对象名称禁止使用mysql保留关键字（如果表名中包含关键字查询时，需要将其用单引号括起来） 数据库对象的命名要能做到见名识意，并且最后不要超过32个字符 临时库表必须以tmp_为前缀并以日期为后缀，备份
阅读更多...
搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
业务中14个需要进行A/B测试的时刻[信息图]

业务中14个需要进行A/B测试的时刻[信息图]

在本指南中，我们将全面了解有关 A/B测试 的所有内容。 我们将介绍不同类型的A/B测试，如何有效地规划和启动测试，如何评估测试是否成功，您应该关注哪些指标，多年来我们发现的常见错误等等。 什么是A/B测试？ A/B测试（有时称为“分割测试”）是一种实验类型，其中您创建两种或多种内容变体——如登录页面、电子邮件或广告——并将它们显示给不同的受众群体，以查看哪一种效果最好。 本质上，A/B测
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2