写点什么

应用实践 | Apache Doris 整合 Iceberg + Flink CDC 构建实时湖仓一体的联邦查询分析架构

作者:SelectDB
  • 2022 年 6 月 23 日
  • 本文字数:8438 字

    阅读完需:约 28 分钟

应用实践 | 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 软件环境

本教程的演示环境如下:


  1. Centos7

  2. Apahce doris 1.1

  3. Hadoop 3.3.3

  4. hive 3.1.3

  5. Fink 1.14.4

  6. flink-sql-connector-mysql-cdc-2.2.1

  7. Apache Iceberg 0.13.2

  8. JDK 1.8.0_311

  9. 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 系统架构

我们整理架构图如下



  1. 首先我们从 Mysql 数据中使用 Flink 通过 Binlog 完成数据的实时采集

  2. 然后再 Flink 中创建 Iceberg 表,Iceberg 的元数据保存在 hive 里

  3. 最后我们在 Doris 中创建 Iceberg 外表

  4. 在通过 Doris 统一查询入口完成对 Iceberg 里的数据进行查询分析,供前端应用调用,这里 iceberg 外表的数据可以和 Doris 内部数据或者 Doris 其他外部数据源的数据进行关联查询分析


Doris 湖仓一体的联邦查询架构如下:



  1. Doris 通过 ODBC 方式支持:MySQL,Postgresql,Oracle ,SQLServer

  2. 同时支持 Elasticsearch 外表

  3. 1.0 版本支持 Hive 外表

  4. 1.1 版本支持 Iceberg 外表

  5. 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

sbin/start-all.sh
复制代码

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

bin/start-cluster.sh
复制代码


启动后的界面如下:


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 表的数据分析问题:


  1. 支持 Iceberg 数据源接入 Doris

  2. 支持 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 注意事项

  • Iceberg 表 Schema 变更不会自动同步,需要在 Doris 中通过 REFRESH 命令同步 Iceberg 外表或数据库。

  • 当前默认支持的 Iceberg 版本为 0.12.0,0.13.x,未在其他版本进行测试。后续后支持更多版本。

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



用户头像

SelectDB

关注

极速易用 开源开放 2022.04.20 加入

The enterprise-grade cloud-native distribution for Apache Doris

评论

发布
暂无评论
应用实践 | Apache Doris 整合 Iceberg + Flink CDC 构建实时湖仓一体的联邦查询分析架构_数据库_SelectDB_InfoQ写作社区