应用实践 | Apache Doris 整合 Iceberg + Flink CDC 构建实时湖仓一体的联邦查询分析架构
导读：这是一篇非常完整全面的应用技术干货，手把手教你如何使用 Doris+Iceberg+Flink CDC 构建实时湖仓一体的联邦查询分析架构。按照本文中步骤一步步完成，完整体验搭建操作的完整过程。
作者｜Apache Doris PMC 成员 张家锋
1.概览
这篇教程将展示如何使用 Doris+Iceberg+Flink CDC 构建实时湖仓一体的联邦查询分析，Doris 1.1 版本提供了 Iceberg 的支持，本文主要展示 Doris 和 Iceberg 怎么使用，同时本教程整个环境是都基于伪分布式环境搭建，大家按照步骤可以一步步完成。完整体验整个搭建操作的过程。
1.1 软件环境
本教程的演示环境如下：
Centos7
Apahce doris 1.1
Hadoop 3.3.3
hive 3.1.3
Fink 1.14.4
flink-sql-connector-mysql-cdc-2.2.1
Apache Iceberg 0.13.2
JDK 1.8.0_311
MySQL 8.0.29
wget https://archive.apache.org/dist/hadoop/core/hadoop-3.3.3/hadoop-3.3.3.tar.gzwget https://archive.apache.org/dist/hive/hive-3.1.3/apache-hive-3.1.3-bin.tar.gzwget https://dlcdn.apache.org/flink/flink-1.14.4/flink-1.14.4-bin-scala_2.12.tgzwget https://search.maven.org/remotecontent?filepath=org/apache/iceberg/iceberg-flink-runtime-1.14/0.13.2/iceberg-flink-runtime-1.14-0.13.2.jarwget https://repository.cloudera.com/artifactory/cloudera-repos/org/apache/flink/flink-shaded-hadoop-3-uber/3.1.1.7.2.9.0-173-9.0/flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar
复制代码
1.2 系统架构
我们整理架构图如下
首先我们从 Mysql 数据中使用 Flink 通过 Binlog 完成数据的实时采集
然后再 Flink 中创建 Iceberg 表，Iceberg 的元数据保存在 hive 里
最后我们在 Doris 中创建 Iceberg 外表
在通过 Doris 统一查询入口完成对 Iceberg 里的数据进行查询分析，供前端应用调用，这里 iceberg 外表的数据可以和 Doris 内部数据或者 Doris 其他外部数据源的数据进行关联查询分析
Doris 湖仓一体的联邦查询架构如下：
Doris 通过 ODBC 方式支持：MySQL，Postgresql，Oracle ，SQLServer
同时支持 Elasticsearch 外表
1.0 版本支持 Hive 外表
1.1 版本支持 Iceberg 外表
1.2 版本支持 Hudi 外表
2.环境安装部署
2.1 安装 Hadoop、Hive
tar zxvf hadoop-3.3.3.tar.gztar zxvf apache-hive-3.1.3-bin.tar.gz
复制代码
配置系统环境变量
export HADOOP_HOME=/data/hadoop-3.3.3export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoopexport HADOOP_HDFS_HOME=$HADOOP_HOMEexport HIVE_HOME=/data/hive-3.1.3export PATH=$PATH:$HADOOP_HOME/bin:$HIVE_HOME/bin:$HIVE_HOME/conf
复制代码
2.2 配置 hdfs
2.2.1 core-site.xml
vi etc/hadoop/core-site.xml
<configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property></configuration>
复制代码
2.2.2 hdfs-site.xml
vi etc/hadoop/hdfs-site.xml
<configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>/data/hdfs/namenode</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>/data/hdfs/datanode</value> </property> </configuration>
复制代码
2.2.3 修改 Hadoop 启动脚本
sbin/start-dfs.sh
sbin/stop-dfs.sh
在文件开始加上下面的内容
HDFS_DATANODE_USER=rootHADOOP_SECURE_DN_USER=hdfsHDFS_NAMENODE_USER=rootHDFS_SECONDARYNAMENODE_USER=root
复制代码
sbin/start-yarn.sh
sbin/stop-yarn.sh
在文件开始加上下面的内容
YARN_RESOURCEMANAGER_USER=rootHADOOP_SECURE_DN_USER=yarnYARN_NODEMANAGER_USER=root
复制代码
2.3 配置 yarn
这里我改变了 Yarn 的一些端口，因为我是单机环境和 Doris 的一些端口冲突。你可以不启动 yarn
vi etc/hadoop/yarn-site.xml
<property> <name>yarn.resourcemanager.address</name> <value>jiafeng-test:50056</value> </property> <property> <name>yarn.resourcemanager.scheduler.address</name> <value>jiafeng-test:50057</value> </property> <property> <name>yarn.resourcemanager.resource-tracker.address</name> <value>jiafeng-test:50058</value> </property> <property> <name>yarn.resourcemanager.admin.address</name> <value>jiafeng-test:50059</value> </property> <property> <name>yarn.resourcemanager.webapp.address</name> <value>jiafeng-test:9090</value> </property> <property> <name>yarn.nodemanager.localizer.address</name> <value>0.0.0.0:50060</value> </property> <property> <name>yarn.nodemanager.webapp.address</name> <value>0.0.0.0:50062</value> </property>
复制代码
vi etc/hadoop/mapred-site.xm
<property> <name>mapreduce.jobhistory.address</name> <value>0.0.0.0:10020</value> </property> <property> <name>mapreduce.jobhistory.webapp.address</name> <value>0.0.0.0:19888</value> </property> <property> <name>mapreduce.shuffle.port</name> <value>50061</value> </property>
复制代码
2.2.4 启动 hadoop
2.4 配置 Hive
2.4.1 创建 hdfs 目录
hdfs dfs -mkdir -p /user/hive/warehousehdfs dfs -mkdir /tmphdfs dfs -chmod g+w /user/hive/warehousehdfs dfs -chmod g+w /tmp
复制代码
2.4.2 配置 hive-site.xml
<?xml version="1.0"?><?xml-stylesheet type="text/xsl" href="configuration.xsl"?><configuration> <property> <name>javax.jdo.option.ConnectionURL</name> <value>jdbc:mysql://localhost:3306/hive?createDatabaseIfNotExist=true</value> </property> <property> <name>javax.jdo.option.ConnectionDriverName</name> <value>com.mysql.jdbc.Driver</value> </property> <property> <name>javax.jdo.option.ConnectionUserName</name> <value>root</value> </property> <property> <name>javax.jdo.option.ConnectionPassword</name> <value>MyNewPass4!</value> </property> <property> <name>hive.metastore.warehouse.dir</name> <value>/user/hive/warehouse</value> <description>location of default database for the warehouse</description> </property> <property> <name>hive.metastore.uris</name> <value/> <description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore.</description> </property> <property> <name>javax.jdo.PersistenceManagerFactoryClass</name> <value>org.datanucleus.api.jdo.JDOPersistenceManagerFactory</value> </property> <property> <name>hive.metastore.schema.verification</name> <value>false</value> </property> <property> <name>datanucleus.schema.autoCreateAll</name> <value>true</value> </property></configuration>
复制代码
2.4.3 配置 hive-env.sh
加入以下内容
HADOOP_HOME=/data/hadoop-3.3.3
复制代码
2.4.4 hive 元数据初始化
schematool -initSchema -dbType mysql
复制代码
2.4.5 启动 hive metaservice
后台运行
nohup bin/hive --service metaservice 1>/dev/null 2>&1 &
复制代码
验证
lsof -i:9083COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAMEjava 20700 root 567u IPv6 54605348 0t0 TCP *:emc-pp-mgmtsvc (LISTEN)
复制代码
2.5 安装 MySQL
具体请参照这里：
使用 Flink CDC 实现 MySQL 数据实时入 Apache Doris
2.5.1 创建 MySQL 数据库表并初始化数据
CREATE DATABASE demo;USE demo;CREATE TABLE userinfo ( id int NOT NULL AUTO_INCREMENT, name VARCHAR(255) NOT NULL DEFAULT 'flink', address VARCHAR(1024), phone_number VARCHAR(512), email VARCHAR(255), PRIMARY KEY (`id`))ENGINE=InnoDB ;INSERT INTO userinfo VALUES (10001,'user_110','Shanghai','13347420870', NULL);INSERT INTO userinfo VALUES (10002,'user_111','xian','13347420870', NULL);INSERT INTO userinfo VALUES (10003,'user_112','beijing','13347420870', NULL);INSERT INTO userinfo VALUES (10004,'user_113','shenzheng','13347420870', NULL);INSERT INTO userinfo VALUES (10005,'user_114','hangzhou','13347420870', NULL);INSERT INTO userinfo VALUES (10006,'user_115','guizhou','13347420870', NULL);INSERT INTO userinfo VALUES (10007,'user_116','chengdu','13347420870', NULL);INSERT INTO userinfo VALUES (10008,'user_117','guangzhou','13347420870', NULL);INSERT INTO userinfo VALUES (10009,'user_118','xian','13347420870', NULL);
复制代码
2.6 安装 Flink
tar zxvf flink-1.14.4-bin-scala_2.12.tgz
复制代码
然后需要将下面的依赖拷贝到 Flink 安装目录下的 lib 目录下，具体的依赖的 lib 文件如下：
下面将几个 Hadoop 和 Flink 里没有的依赖下载地址放在下面
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/flink-sql-connector-mysql-cdc-2.2.1.jarwget https://repo1.maven.org/maven2/org/apache/thrift/libfb303/0.9.3/libfb303-0.9.3.jarwget https://search.maven.org/remotecontent?filepath=org/apache/iceberg/iceberg-flink-runtime-1.14/0.13.2/iceberg-flink-runtime-1.14-0.13.2.jarwget https://repository.cloudera.com/artifactory/cloudera-repos/org/apache/flink/flink-shaded-hadoop-3-uber/3.1.1.7.2.9.0-173-9.0/flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar
复制代码
其他的：
hadoop-3.3.3/share/hadoop/common/lib/commons-configuration2-2.1.1.jarhadoop-3.3.3/share/hadoop/common/lib/commons-logging-1.1.3.jarhadoop-3.3.3/share/hadoop/tools/lib/hadoop-archive-logs-3.3.3.jarhadoop-3.3.3/share/hadoop/common/lib/hadoop-auth-3.3.3.jarhadoop-3.3.3/share/hadoop/common/lib/hadoop-annotations-3.3.3.jarhadoop-3.3.3/share/hadoop/common/hadoop-common-3.3.3.jaradoop-3.3.3/share/hadoop/hdfs/hadoop-hdfs-3.3.3.jarhadoop-3.3.3/share/hadoop/client/hadoop-client-api-3.3.3.jarhive-3.1.3/lib/hive-exec-3.1.3.jarhive-3.1.3/lib/hive-metastore-3.1.3.jarhive-3.1.3/lib/hive-hcatalog-core-3.1.3.jar
复制代码
2.6.1 启动 Flink
启动后的界面如下：
2.6.2 进入 Flink SQL Client
bin/sql-client.sh embedded
复制代码
开启 checkpoint，每隔 3 秒做一次 checkpoint
Checkpoint 默认是不开启的，我们需要开启 Checkpoint 来让 Iceberg 可以提交事务。 并且，mysql-cdc 在 binlog 读取阶段开始前，需要等待一个完整的 checkpoint 来避免 binlog 记录乱序的情况。
注意：
这里是演示环境，checkpoint 的间隔设置比较短，线上使用，建议设置为 3-5 分钟一次 checkpoint。
Flink SQL> SET execution.checkpointing.interval = 3s;[INFO] Session property has been set.
复制代码
2.6.3 创建 Iceberg Catalog
CREATE CATALOG hive_catalog WITH ( 'type'='iceberg', 'catalog-type'='hive', 'uri'='thrift://localhost:9083', 'clients'='5', 'property-version'='1', 'warehouse'='hdfs://localhost:8020/user/hive/warehouse');
复制代码
查看 catalog
Flink SQL> show catalogs;+-----------------+| catalog name |+-----------------+| default_catalog || hive_catalog |+-----------------+2 rows in set
复制代码
2.6.4 创建 Mysql CDC 表
CREATE TABLE user_source ( database_name STRING METADATA VIRTUAL, table_name STRING METADATA VIRTUAL, `id` DECIMAL(20, 0) NOT NULL, name STRING, address STRING, phone_number STRING, email STRING, PRIMARY KEY (`id`) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'root', 'password' = 'MyNewPass4!', 'database-name' = 'demo', 'table-name' = 'userinfo' );
复制代码
查询 CDC 表:
select * from user_source;
复制代码
2.6.5 创建 Iceberg 表
---查看catalogshow catalogs;---使用cataloguse catalog hive_catalog;--创建数据库CREATE DATABASE iceberg_hive; --使用数据库use iceberg_hive;
复制代码
2.6.5.1 创建表
CREATE TABLE all_users_info ( database_name STRING, table_name STRING, `id` DECIMAL(20, 0) NOT NULL, name STRING, address STRING, phone_number STRING, email STRING, PRIMARY KEY (database_name, table_name, `id`) NOT ENFORCED ) WITH ( 'catalog-type'='hive' );
复制代码
从 CDC 表里插入数据到 Iceberg 表里
use catalog default_catalog;insert into hive_catalog.iceberg_hive.all_users_info select * from user_source;
复制代码
在 web 界面可以看到任务的运行情况
然后停掉任务，我们去查询 iceberg 表
select * from hive_catalog.iceberg_hive.all_users_info
复制代码
可以看到下面的结果
我们去 hdfs 上可以看到 hive 目录下的数据及对应的元数据
我们也可以通过 Hive 建好 Iceberg 表，然后通过 Flink 将数据插入到表里
下载 Iceberg Hive 运行依赖
wget https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-hive-runtime/0.13.2/iceberg-hive-runtime-0.13.2.jar
复制代码
在 hive shell 下执行：
SET engine.hive.enabled=true; SET iceberg.engine.hive.enabled=true; SET iceberg.mr.catalog=hive; add jar /path/to/iiceberg-hive-runtime-0.13.2.jar;
复制代码
创建表
CREATE EXTERNAL TABLE iceberg_hive( `id` int, `name` string)STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler' LOCATION 'hdfs://localhost:8020/user/hive/warehouse/iceber_db/iceberg_hive'TBLPROPERTIES ( 'iceberg.mr.catalog'='hadoop', 'iceberg.mr.catalog.hadoop.warehouse.location'='hdfs://localhost:8020/user/hive/warehouse/iceber_db/iceberg_hive' );
复制代码
然后再 Flink SQL Client 下执行下面语句将数据插入到 Iceber 表里
INSERT INTO hive_catalog.iceberg_hive.iceberg_hive values(2, 'c');INSERT INTO hive_catalog.iceberg_hive.iceberg_hive values(3, 'zhangfeng');
复制代码
查询这个表
select * from hive_catalog.iceberg_hive.iceberg_hive
复制代码
可以看到下面的结果
3. Doris 查询 Iceberg
Apache Doris 提供了 Doris 直接访问 Iceberg 外部表的能力，外部表省去了繁琐的数据导入工作，并借助 Doris 本身的 OLAP 的能力来解决 Iceberg 表的数据分析问题：
支持 Iceberg 数据源接入 Doris
支持 Doris 与 Iceberg 数据源中的表联合查询，进行更加复杂的分析操作
3.1 安装 Doris
这里我们不在详细讲解 Doris 的安装，如果你不知道怎么安装 Doris 请参照官方文档：快速入门
3.2 创建 Iceberg 外表
CREATE TABLE `all_users_info` ENGINE = ICEBERGPROPERTIES ("iceberg.database" = "iceberg_hive","iceberg.table" = "all_users_info","iceberg.hive.metastore.uris" = "thrift://localhost:9083","iceberg.catalog.type" = "HIVE_CATALOG");
复制代码
参数说明：
ENGINE 需要指定为 ICEBERG
PROPERTIES 属性：
iceberg.hive.metastore.uris：Hive Metastore 服务地址
iceberg.database：挂载 Iceberg 对应的数据库名
iceberg.table：挂载 Iceberg 对应的表名，挂载 Iceberg database 时无需指定。
iceberg.catalog.type：Iceberg 中使用的 catalog 方式，默认为 HIVE_CATALOG，当前仅支持该方式，后续会支持更多的 Iceberg catalog 接入方式。
mysql> CREATE TABLE `all_users_info` -> ENGINE = ICEBERG -> PROPERTIES ( -> "iceberg.database" = "iceberg_hive", -> "iceberg.table" = "all_users_info", -> "iceberg.hive.metastore.uris" = "thrift://localhost:9083", -> "iceberg.catalog.type" = "HIVE_CATALOG" -> );Query OK, 0 rows affected (0.23 sec)mysql> select * from all_users_info;+---------------+------------+-------+----------+-----------+--------------+-------+| database_name | table_name | id | name | address | phone_number | email |+---------------+------------+-------+----------+-----------+--------------+-------+| demo | userinfo | 10004 | user_113 | shenzheng | 13347420870 | NULL || demo | userinfo | 10005 | user_114 | hangzhou | 13347420870 | NULL || demo | userinfo | 10002 | user_111 | xian | 13347420870 | NULL || demo | userinfo | 10003 | user_112 | beijing | 13347420870 | NULL || demo | userinfo | 10001 | user_110 | Shanghai | 13347420870 | NULL || demo | userinfo | 10008 | user_117 | guangzhou | 13347420870 | NULL || demo | userinfo | 10009 | user_118 | xian | 13347420870 | NULL || demo | userinfo | 10006 | user_115 | guizhou | 13347420870 | NULL || demo | userinfo | 10007 | user_116 | chengdu | 13347420870 | NULL |+---------------+------------+-------+----------+-----------+--------------+-------+9 rows in set (0.18 sec)
复制代码
3.3 同步挂载
当 Iceberg 表 Schema 发生变更时，可以通过 REFRESH 命令手动同步，该命令会将 Doris 中的 Iceberg 外表删除重建。
-- 同步 Iceberg 表REFRESH TABLE t_iceberg;-- 同步 Iceberg 数据库REFRESH DATABASE iceberg_test_db;
复制代码
3.4 Doris 和 Iceberg 数据类型对应关系
支持的 Iceberg 列类型与 Doris 对应关系如下表：
3.5 注意事项
3.6 Doris FE 配置
下面几个配置属于 Iceberg 外表系统级别的配置，可以通过修改 fe.conf 来配置，也可以通过 ADMIN SET CONFIG 来配置。
iceberg_table_creation_strict_mode
创建 Iceberg 表默认开启 strict mode。 strict mode 是指对 Iceberg 表的列类型进行严格过滤，如果有 Doris 目前不支持的数据类型，则创建外表失败。
iceberg_table_creation_interval_second
自动创建 Iceberg 表的后台任务执行间隔，默认为 10s。
max_iceberg_table_creation_record_size
Iceberg 表创建记录保留的最大值，默认为 2000. 仅针对创建 Iceberg 数据库记录。
4. 总结
这里 Doris On Iceberg 我们只演示了 Iceberg 单表的查询，你还可以联合 Doris 的表，或者其他的 ODBC 外表，Hive 外表，ES 外表等进行联合查询分析，通过 Doris 对外提供统一的查询分析入口。
自此我们完整从搭建 Hadoop，hive、flink 、Mysql、Doris 及 Doris On Iceberg 的使用全部介绍完了，Doris 朝着数据仓库和数据融合的架构演进，支持湖仓一体的联邦查询，给我们的开发带来更多的便利，更高效的开发，省去了很多数据同步的繁琐工作，快快来体验吧。最后，欢迎更多的开源技术爱好者加入 Apache Doris 社区，携手成长，共建社区生态。
SelectDB 是一家开源技术公司，致力于为 Apache Doris 社区提供一个由全职工程师、产品经理和支持工程师组成的团队，繁荣开源社区生态，打造实时分析型数据库领域的国际工业界标准。基于 Apache Doris 研发的新一代云原生实时数仓 SelectDB，运行于多家云上，为用户和客户提供开箱即用的能力。
相关链接：
SelectDB 官方网站：
https://selectdb.com (We Are Coming Soon)
Apache Doris 官方网站：
http://doris.apache.org
Apache Doris Github：
https://github.com/apache/doris
Apache Doris 开发者邮件组：
dev@doris.apache.org
评论