Spring Aware机制

李鸿坤

发布于 2020-07-18 10:39:24

1K00

代码可运行

文章被收录于专栏：泛泛聊后端泛泛聊后端

运行总次数：0

代码可运行

流计算 Oceanus 简介

流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。

本文将您详细介绍如何提取 MySQL 数据与 HBase 数据进行维表关联（流维 join），经过简单聚合分析后存入 Elasticsearch 中。

前置准备

创建流计算 Oceanus 集群

在流计算 Oceanus 产品活动页面 1 元购买 Oceanus 集群。

进入 Oceanus 控制台 [1]，点击左侧【集群管理】，点击左上方【创建集群】，具体可参考 Oceanus 官方文档 创建独享集群 [2]。

创建 MySQL 实例

进入 MySQL 控制台 [3]，点击【新建】。具体可参考官方文档 创建 MySQL 实例 [4]。

数据准备：

进入实例数据库，创建 oceanus_advanced1_student_grade 表，并手动插入数据。

-- 建表语句
CREATE TABLE `oceanus_advanced1_student_grade` (
  `name`    varchar(50) NOT NULL DEFAULT '',
  `grade`   int(3)               DEFAULT NULL,
  PRIMARY KEY (`name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
-- 数据插入
INSERT INTO `oceanus_advanced1_student_grade` (`name`, `grade`) VALUES ('Oceanus-1', 85);
INSERT INTO `oceanus_advanced1_student_grade` (`name`, `grade`) VALUES ('Oceanus-2', 95)

创建 EMR 集群

数据准备：

# 进入 HBase 命令
root@yourhostname~# hbase shell

-- 建表语句
create 'oceanus_advanced1_student_info','StuInfo'
-- 数据插入
put 'oceanus_advanced1_student_info','Oceanus-1','StuInfo:Class','01'
put 'oceanus_advanced1_student_info','Oceanus-1','StuInfo:Age','17'
put 'oceanus_advanced1_student_info','Oceanus-2','StuInfo:Class','01'
put 'oceanus_advanced1_student_info','Oceanus-2','StuInfo:Age','20'
put 'oceanus_advanced1_student_info','Oceanus-3','StuInfo:Class','01'
put 'oceanus_advanced1_student_info','Oceanus-3','StuInfo:Age','18'

创建 Elasticsearch 集群

进入 Elasticsearch 控制台 [7]，点击左上方【新建】，创建 Elasticsearch 实例，具体操作请访问 创建 Elasticsearch 集群 [8]

流计算 Oceanus 作业

1. 创建 Source

CREATE TABLE `mysql_cdc_source_table` (
  `name`        STRING,
  `grade`       STRING,
  `proc_time`   AS PROCTIME(),      -- 这里 proc_time 字段配合下面流维 join 时使用。
  PRIMARY KEY (`name`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
  'connector' = 'mysql-cdc',        -- 固定值 'mysql-cdc'
  'hostname' = '10.0.0.158',        -- 数据库的 IP
  'port' = '3306',                  -- 数据库的访问端口
  'username' = 'root',              -- 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限）
  'password' = 'Tencent123$',       -- 数据库访问的密码
  'database-name' = 'testdb',       -- 需要同步的数据库
  'table-name' = 'oceanus_advanced1_student_grade'   -- 需要同步的数据表名
);

2. 创建 HBase 维表

CREATE TABLE hbase_table (
  rowkey      STRING,
  StuInfo     ROW <Class STRING,Age STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'hbase-1.4',                         -- Flink 1.13 支持 hbase-2.2
  'table-name' = 'oceanus_advanced1_student_info',   -- HBase 表名
  'zookeeper.quorum' = '10.0.0.118:2181,10.0.0.119:2181,10.0.0.3:2181'   -- HBase 的 zookeeper 地址
);

3. 创建 Sink

CREATE TABLE elasticsearch6_sink_table (
    `class`       STRING,
    `amount`      BIGINT,
    PRIMARY KEY(`class`) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-6',      -- 输出到 Elasticsearch 6
    'username' = 'elastic',               -- 选填 用户名
    'password' = 'Tencent123$',           -- 选填 密码
    'hosts' = 'http://10.0.0.97:9200',    -- Elasticsearch 的连接地址
    'index' = 'oceanus_advanced1',        -- Elasticsearch 的 Index 名
    'document-type' = '_doc',             -- Elasticsearch 的 Document 类型
    'sink.bulk-flush.max-actions' = '1',  -- 每条数据都刷新
    'format' = 'json'                     -- 输出数据格式，目前只支持 'json'
);

4. 编写业务 SQL

INSERT INTO elasticsearch6_sink_table
SELECT
b.StuInfo.Class            AS class,
COUNT(a.name)              AS amount
FROM mysql_cdc_source_table AS a
JOIN hbase_table FOR SYSTEM_TIME AS OF a.proc_time AS b
-- 这里一定要加入 for SYSTEM_TIME as of 语句，否则虽然仍然可以执行 JOIN，但是只会全量读取一次数据库，结果可能不符合预期。
ON a.name = b.rowkey
WHERE CAST(a.grade AS INT) >= 90 AND CAST(b.StuInfo.Age AS INT) >= 18
GROUP BY b.StuInfo.Class

总结

流计算 Oceanus 平台当前内置 Connector 支持了 MySQL、PostgreSQL、Hive、HBase、Redis 和 Oracle 维表，无需用户自己开发即可使用。具体如何使用维表可参考 内置维表参考列表 [9] 本例统计的是各个班级年龄大于等于18岁，并且成绩大于等于90分的人数，无实际业务用途。