简单建优化网站无需技术,成都网站优化平台,网站建设管理职责,carwling wordpress简介#xff1a;在实际业务使用中#xff0c;需要经常实时做一些数据分析#xff0c;包括实时PV和UV展示#xff0c;实时销售数据#xff0c;实时店铺UV以及实时推荐系统等#xff0c;基于此类需求#xff0c;Confluent实时计算Flink版是一个高效的方案。
业务背景
在…简介在实际业务使用中需要经常实时做一些数据分析包括实时PV和UV展示实时销售数据实时店铺UV以及实时推荐系统等基于此类需求Confluent实时计算Flink版是一个高效的方案。
业务背景
在实际业务使用中需要经常实时做一些数据分析包括实时PV和UV展示实时销售数据实时店铺UV以及实时推荐系统等基于此类需求Confluent实时计算Flink版是一个高效的方案。
Confluent是基于Apache Kafka提供的企业级全托管流数据服务由 Apache Kafka 的原始创建者构建通过企业级功能扩展了 Kafka 的优势同时消除了 Kafka管理或监控的负担。
实时计算Flink版是阿里云基于 Apache Flink 构建的企业级实时大数据计算商业产品。实时计算 Flink 由 Apache Flink 创始团队官方出品拥有全球统一商业化品牌提供全系列产品矩阵完全兼容开源 Flink API并充分基于强大的阿里云平台提供云原生的 Flink 商业增值能力。
一、准备工作-创建Confluent集群和实时计算Flink版集群
登录Confluent管理控制台创建Confluent集群创建步骤参考 Confluent集群开通
登录实时计算Flink版管理控制台创建vvp集群。请注意创建vvp集群选择的vpc跟confluent集群的region和vpc使用同一个这样可以在vvp内部访问confluent的内部域名。
二、最佳实践-实时统计玩家充值金额-Confluent实时计算FlinkHologres
2.1 新建Confluent消息队列
在confluent集群列表页登录control center在左侧选中Topics点击Add a topic按钮创建一个名为confluent-vvp-test的topic将partition设置为3
2.2 配置结果表 Hologres
进入Hologres控制台点击Hologres实例在DB管理中新增数据库mydb登录Hologres数据库新建SQLHologres中创建结果表 SQL语句
--用户累计消费结果表CREATE TABLE consume (appkey VARCHAR,serverid VARCHAR,servertime VARCHAR,roleid VARCHAR,amount FLOAT,dt VARCHAR,primary key(appkey,dt));
2.3 创建实时计算vvp作业
首先登录vvp控制台选择集群所在region点击控制台进入开发界面点击作业开发Tab点击新建文件文件名称confluent-vvp-hologres文件类型选择流作业/SQL在输入框写入以下代码
create TEMPORARY table kafka_game_consume_source( appkey STRING,servertime STRING,consumenum DOUBLE,roleid STRING,serverid STRING
) with (connector kafka,topic game_consume_log,properties.bootstrap.servers kafka.confluent.svc.cluster.local.xxx:9071[xxx可以找开发同学查看],properties.group.id gamegroup,format json,properties.ssl.truststore.location /flink/usrlib/truststore.jks,properties.ssl.truststore.password [your truststore password],properties.security.protocolSASL_SSL,properties.sasl.mechanismPLAIN,properties.sasl.jaas.configorg.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required usernamexxx[集群的用户] passwordxxx[相应的密码];
);
-- 创建累计消费hologres sink表
CREATE TEMPORARY TABLE consume(appkey STRING,serverid STRING,servertime STRING,roleid STRING,amount DOUBLE,dt STRING,PRIMARY KEY (appkey,dt) NOT ENFORCED)WITH (connector hologres,dbname mydb,endpoint hgprecn-cn-tl32gkaet006-cn-beijing-vpc.hologres.aliyuncs.com:80,password [your appkey secret],tablename consume,username [your app key],mutateType insertorreplace);
--{appkey:appkey1,servertime:2020-09-30 14:10:36,consumenum:33.8,roleid:roleid1,serverid:1}
--{appkey:appkey2,servertime:2020-09-30 14:11:36,consumenum:30.8,roleid:roleid2,serverid:2}
--{appkey:appkey1,servertime:2020-09-30 14:13:36,consumenum:31.8,roleid:roleid1,serverid:1}
--{appkey:appkey2,servertime:2020-09-30 14:20:36,consumenum:33.8,roleid:roleid2,serverid:2}
--{appkey:appkey1,servertime:2020-09-30 14:30:36,consumenum:73.8,roleid:roleid1,serverid:1}-- 计算每个用户累积消费金额insert into consumeSELECTappkey,LAST_VALUE(serverid) as serverid,LAST_VALUE(servertime) as servertime,LAST_VALUE(roleid) as roleid,sum(consumenum) as amount,substring(servertime,1,10) as dtFROM kafka_game_consume_sourceGROUP BY appkey,substring(servertime,1,10)having sum(consumenum) 0;
在高级配置里增加依赖文件truststore.jks访问内部域名得添加这个文件访问公网域名可以不用访问依赖文件的固定路径前缀都是/flink/usrlib/这里就是/flink/usrlib/truststore.jks点击上线按钮完成上线在运维作用列表里找到刚上线的作用点击启动按钮等待状态更新为running运行成功。在control center的【Topics-Messages】页面逐条发送测试消息格式为
{appkey:appkey1,servertime:2020-09-30 14:10:36,consumenum:33.8,roleid:roleid1,serverid:1}
{appkey:appkey2,servertime:2020-09-30 14:11:36,consumenum:30.8,roleid:roleid2,serverid:2}
{appkey:appkey1,servertime:2020-09-30 14:13:36,consumenum:31.8,roleid:roleid1,serverid:1}
{appkey:appkey2,servertime:2020-09-30 14:20:36,consumenum:33.8,roleid:roleid2,serverid:2}
{appkey:appkey1,servertime:2020-09-30 14:30:36,consumenum:73.8,roleid:roleid1,serverid:1}
2.4 查看用户充值金额实时统计效果 三、最佳实践-电商实时PV和UV统计-Confluent实时计算FlinkRDS
3.1 新建Confluent消息队列
在confluent集群列表页登录control center在左侧选中Topics点击Add a topic按钮创建一个名为pv-uv的topic将partition设置为3
3.2 创建云数据库RDS结果表
登录 RDS 管理控制台页面购买RDS。确保RDS与Flink全托管集群在相同region相同VPC下添加虚拟交换机网段vswitch IP段进入RDS白名单详情参考设置白名单文档3.【vswitch IP段】可在 flink的工作空间详情中查询 在【账号管理】页面创建账号【高权限账号】数据库实例下【数据库管理】新建数据库【conflufent_vvp】使用系统自带的DMS服务登陆RDS登录名和密码输入上面创建的高权限账户双击【confluent_vvp】数据库打开SQLConsole将以下建表语句复制粘贴到 SQLConsole中创建结果表
CREATE TABLE result_cps_total_summary_pvuv_min(summary_date date NOT NULL COMMENT 统计日期,summary_min varchar(255) COMMENT 统计分钟,pv bigint COMMENT pv,uv bigint COMMENT uv,currenttime timestamp COMMENT 当前时间,primary key(summary_date,summary_min)
) 3.3 创建实时计算VVP作业
1.【[VVP控制台】新建文件 在SQL区域输入以下代码
--数据的订单源表
CREATE TABLE source_ods_fact_log_track_action (account_id VARCHAR,--用户IDclient_ip VARCHAR,--客户端IPclient_info VARCHAR,--设备机型信息platform VARCHAR,--系统版本信息imei VARCHAR,--设备唯一标识version VARCHAR,--版本号action VARCHAR,--页面跳转描述gpm VARCHAR,--埋点链路c_time VARCHAR,--请求时间target_type VARCHAR,--目标类型target_id VARCHAR,--目标IDudata VARCHAR,--扩展信息JSON格式session_id VARCHAR,--会话IDproduct_id_chain VARCHAR,--商品ID串cart_product_id_chain VARCHAR,--加购商品IDtag VARCHAR,--特殊标记position VARCHAR,--位置信息network VARCHAR,--网络使用情况p_dt VARCHAR,--时间分区天p_platform VARCHAR --系统版本信息
) WITH (connector kafka,topic game_consume_log,properties.bootstrap.servers kafka.confluent.svc.cluster.local.c79f69095bc5d4d98b01136fe43e31b93:9071,properties.group.id gamegroup,format json,properties.ssl.truststore.location /flink/usrlib/truststore.jks,properties.ssl.truststore.password 【your password】,properties.security.protocolSASL_SSL,properties.sasl.mechanismPLAIN,properties.sasl.jaas.configorg.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username【your user name】 password【your password】;
);
--{account_id:id1,client_ip:172.11.1.1,client_info:mi10,p_dt:2021-12-01,c_time:2021-12-01 19:10:00}
CREATE TABLE result_cps_total_summary_pvuv_min (summary_date date,--统计日期summary_min varchar,--统计分钟pv bigint,--点击量uv bigint,--一天内同个访客多次访问仅计算一个UVcurrenttime timestamp,--当前时间primary key (summary_date, summary_min)
) WITH (type rds,url url jdbc:mysql://rm-【your rds clusterId】.mysql.rds.aliyuncs.com:3306/confluent_vvp,,tableName result_cps_total_summary_pvuv_min,userName flink_confluent_vip,password 【your rds password】
);
CREATE VIEW result_cps_total_summary_pvuv_min_01 AS
selectcast (p_dt as date) as summary_date --时间分区, count (client_ip) as pv --客户端的IP, count (distinct client_ip) as uv --客户端去重, cast (max (c_time) as TIMESTAMP) as c_time --请求的时间
fromsource_ods_fact_log_track_action
groupby p_dt;
INSERTinto result_cps_total_summary_pvuv_min
selecta.summary_date,--时间分区cast (DATE_FORMAT (c_time, HH:mm) as varchar) as summary_min,--取出小时分钟级别的时间a.pv,a.uv,CURRENT_TIMESTAMP as currenttime --当前时间
fromresult_cps_total_summary_pvuv_min_01 AS a;
点击【上线】之后在作业运维页面点击启动按钮直到状态更新为RUNNING状态。在control center的【Topics-Messages】页面逐条发送测试消息格式为
{account_id:id1,client_ip:72.11.1.111,client_info:mi10,p_dt:2021-12-01,c_time:2021-12-01 19:11:00}
{account_id:id2,client_ip:72.11.1.112,client_info:mi10,p_dt:2021-12-01,c_time:2021-12-01 19:12:00}
{account_id:id3,client_ip:72.11.1.113,client_info:mi10,p_dt:2021-12-01,c_time:2021-12-01 19:13:00} 3.4 查看PV和UV效果 可以看出rds数据表的pv和uv会随着发送的消息数据动态的变化同时还可以通过【数据可视化】来查看相应的图表信息。 pv图表展示 uv图表展示 原文链接
本文为阿里云原创内容未经允许不得转载。
