Flink+FlinkCDC+Zeppelin单机部署实现FlinkSQL编辑练习

Last updated on 2022年10月19日

Zeppelin是一个很好的多语言，可视化练习的平台，可以很方便的验证我的思路。当然也可以作为BI分析师的好工具，甚至可以当作BI报表工具。

前置条件

• java8环境
• Flink v1.13.3
• FlinkCDC v2.2.1（支持Flink1.13.*, 1.14.*）
• Zeppelin v0.10.1 （同时支持 Flink 1.10 到 1.13 的 4 个大版本，并且同时支持Scala-2.11 和Scala-2.12）

java -version

一、单机部署Flink

下载Flink：Apache Flink: Downloads

以 flink-1.13.3-bin-scala_2.12.tgz 为例

mkdir flink && cd flink
wget https://archive.apache.org/dist/flink/flink-1.13.3/flink-1.13.3-bin-scala_2.12.tgz
tar xzf flink-1.13.3-bin-scala_2.12.tgz
cd flink-1.13.3

启动与停止

# 启动本地集群
./bin/start-cluster.sh
# 检查启动情况
ps aux | grep flink

# 停止本地集群
./bin/stop-cluster.sh

二、部署Zeppelin

官网：Zeppelin (apache.org)

部署文档：Zeppelin 搭建 · 语雀 (yuque.com)

部署完成后，配置Flink的HOME变量等

在127.0.0.1:8080/#/interpreter页面搜索flink，然后编辑按钮编辑变量

Name	Value
FLINK_HOME	/path/to/flink-1.13.3
flink.execution.mode	local
flink.execution.jars	/path/to/zeppelin/flink-sql-connector-mysql-cdc-2.2.1.jar

通过Zeppelin提供的变量 flink.execution.jars 引入下载好，对应FlinkCDC对应版本的连接器，上面以mysql为例。

三、启用FlinkSQL编辑练习

新建一个Note，然后执行

%flink.ssql

-- 源表-商消费表
DROP TABLE IF EXISTS source_log;
CREATE TABLE source_log (
    `id` INT NOT NULL
    ,`count` INT NOT NULL
    ,`month` INT NOT NULL COMMENT '月份'
    ,`add_time` TIMESTAMP COMMENT '时间'
    ,PRIMARY KEY ( `id` ) NOT ENFORCED
) COMMENT '记录表'
 WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'abc.mysql.rds.aliyuncs.com',
    'port' = '3306',
    'username' = 'wwwroot',
    'password' = 'testtest',
    'database-name' = 'test',
    'table-name' = 'log'
  );
  
select max(id) from source_log;