Flink 实践教程 - 进阶（1）：维表关联

2021 年 12 月 04 日
本文字数：2822 字
阅读完需：约 9 分钟

作者：腾讯云流计算 Oceanus 团队

流计算 Oceanus 简介

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

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

操作视频

前置准备

创建流计算 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_tableSELECTb.StuInfo.Class            AS class,COUNT(a.name)              AS amountFROM mysql_cdc_source_table AS aJOIN hbase_table FOR SYSTEM_TIME AS OF a.proc_time AS b-- 这里一定要加入 for SYSTEM_TIME as of 语句，否则虽然仍然可以执行 JOIN，但是只会全量读取一次数据库，结果可能不符合预期。ON a.name = b.rowkeyWHERE CAST(a.grade AS INT) >= 90 AND CAST(b.StuInfo.Age AS INT) >= 18GROUP BY b.StuInfo.Class

复制代码