参考资料
| 文档 | 文档链接 |
|---|---|
| 阿里云实时计算flink文档 | https://help.aliyun.com/zh/flink/product-overview/what-is-alibaba-cloud-realtime-compute-for-apache-flink?spm=a2c4g.11186623.help-menu-45029.d_0_0_0.1bbc67b8VuD3An |
| hologres数据库说明文档 | https://help.aliyun.com/zh/hologres/user-guide/instance-types-and-management/?spm=a2c4g.11186623.help-menu-113622.d_2_0.66cce46clYPQL6 |
| flink+hologres落地实践 | https://help.aliyun.com/zh/flink/use-cases/build-real-time-data-warehouse-based-on-flink-hologres?spm=a2c4g.11186623.help-menu-45029.d_3_0_2.59c47d48K0vSaw |
1、为什么是hologres
① flink是国内主流的实时处理工具，那么为何要采用 hologres数据库作为存储呢？
第一、hologres 拥有 binlog的能力。（数据库必须拥有binlog,实时流才能抽取，同时cdcmode 有两种，一种是同步时保持和原表一模一样，另一种是 得到这条记录变化了什么，记录 增删改 操作。可以看到数据库表的变更情况）。
第二、hologres支持行存、列存、行列共存（本质是行存一份数据、列存一份数据） ，这样的话，它就可以 即作为 flink source的维度表，因为是行存储，又可以直接进行查数，分析数据，因为是列存储。
第三、主从实例，读写分离，主实例用来插入和变更，从实例负责读取，这保证了Flink对Hologres Binlog的数据拉取不影响线上服务。
②阿里云实时计算 + hologres 都是阿里云上处理的收费情况如何？
收费还是很昂贵的。主要分两部分
1、flink 1cu/月 180元。（1cu,相当于是1个cpu 4G内存使用1天）
2、hologres 价格=计算+存储，如：每个月需要标准存储500 GB，计算128 CU，总共使用6个月，如果开通的是华东1（杭州）地区，则包年包月需要付的费用为1281701+500*1=22260。
虽然贵，但是也省去了自己购买服务器，和运维人员工资。至于说数据安全，云服务厂商从技术角度，它只要想看，那就有。
2、分层架构
| 实时数仓分层 | 具体过程，从下到上 |
|---|---|
| dws | flink读取dwd的hologres并关联其他表加工聚合到dws的hologres |
| dwd | flink读取ods的hologres,并关联其他表加工到dwd的hologres |
| ods | flink抽取mysql 存储到 hologres |
| 业务系统数据库-mysql |
3、使用案例
1、同步业务数据到ods层
CREATE DATABASE IF NOT EXISTS dw.order_dw -- 创建catalog时设置了table_property.binlog.level参数，因此通过CDAS创建的所有表都开启了binlog。
AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- 可以根据需要选择上游数据库需要入仓的表。
/*+ OPTIONS('server-id'='8001-8004') */ ; -- 指定mysql-cdc实例server-id范围。
2、第二步，配置取数位置 加工dwd层
配置取数位置
CREATE TEMPORARY TABLE source_sqlserver_my_yl_weigh_tmp
(fid int,finspectno varchar(30),ticketno varchar(64),station varchar(30),truckno varchar(30),table_name STRING METADATA FROM 'table_name' VIRTUAL,schema_name STRING METADATA FROM 'schema_name' VIRTUAL,database_name STRING METADATA FROM 'database_name' VIRTUAL,op_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL-- ,op STRING METADATA FROM 'value.op' VIRTUAL,PRIMARY KEY (taredatetime) NOT ENFORCED
)
WITH ('connector' = 'sqlserver-cdc-muyuan','hostname' = '???????' -- 从库,'port' = '?????','database-name' = 'database-name','table-name' = 'xxx.table','username' = '${secret_values.sqlserver_MY_GrainsSystem_username}','password' = '${secret_values.sqlserver_MY_GrainsSystem_password}','scan.startup.mode' = 'latest-offset'
)
;CREATE TEMPORARY TABLE sink_holo_ods_sl_my_yl_weigh
(fid BIGINT,finspectno varchar,ticketno varchar,station varchar,truckno varchar,contractno varchar,PRIMARY KEY (taredatetime) NOT ENFORCED
)
WITH ('connector' = 'hologres','endPoint' = '?????','dbname' = '???','tablename' = '????','username' = '${secret_values.holo_holo_data_gover_test_username_zhz}','password' = '${secret_values.holo_holo_data_gover_test_password_zhz}','mutatetype' = 'insertorreplace','ignoredelete' = 'false'--,'binlog' = 'true'--,'cdcMode' = 'false'--,'binlogStartupMode' = 'initial'
)
;
INSERT INTO sink_holo_ods_sl_my_yl_weigh
SELECTfid,finspectno,ticketno,station,truckno,contractno,product,specification,sender,receiver,transporter
FROM source_sqlserver_my_yl_weigh_tmp
where taredatetime is not null
;
加工dwd层
BEGIN STATEMENT SET;INSERT INTO dw.order_dw.dwd_orders (order_id,order_user_id,order_shop_id,order_product_id,order_fee,order_create_time,order_update_time,order_state,order_product_catalog_name) SELECT o.*, dim.catalog_name FROM dw.order_dw.orders as oLEFT JOIN dw.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dimON o.product_id = dim.product_id;INSERT INTO dw.order_dw.dwd_orders (pay_id, order_id, pay_platform, pay_create_time)SELECT * FROM dw.order_dw.orders_pay;END;
3、加工 dws层
BEGIN STATEMENT SET;INSERT INTO dw.order_dw.dws_usersSELECT order_user_id,DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,SUM (order_fee)FROM dw.order_dw.dwd_orders cWHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 订单流和支付流数据都已写入宽表。GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');INSERT INTO dw.order_dw.dws_shopsSELECT order_shop_id,DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,SUM (order_fee)FROM dw.order_dw.dwd_orders cWHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 订单流和支付流数据都已写入宽表。GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
END;
