Flink datagen代码：

public class TestDataKafka2doris {private  static final String JDBC_SQL =  "CREATE TABLE join_test (\n"+" id INT,\n" +" name STRING\n"+" ) WITH (\n"+"   'connector' = 'jdbc',\n"+"   'url' = 'jdbc:mysql://192.168.6.143:9030/example_db?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC',\n"+"   'driver' = 'com.mysql.jdbc.Driver',\n"+"   'table-name' = 'join_test',\n"+"   'username' = 'root',\n"+"   'password' = 'root'\n"+" )";private static final String DATA_GEN =  "CREATE TABLE datagen (\n" +" id STRING,\n" +" name STRING,\n" +" user_age STRING,\n" +" user_other STRING,\n" +" ts AS localtimestamp\n" +") WITH (\n" +" 'connector' = 'datagen',\n" +" 'rows-per-second'='10',\n" +" 'fields.id.kind'='sequence',\n" +" 'fields.id.start'='1',\n" +" 'fields.id.end'='10000000',\n" +" 'fields.user_age.min'='1',\n" +" 'fields.user_age.max'='1000',\n" +" 'fields.name.length'='2',\n" +" 'fields.user_other.length'='10'\n" +")";private  static final String KAFKA_SQL ="CREATE TABLE kafkaTable (" +" id STRING,\n" +" name STRING,\n" +" user_age STRING,\n" +" user_other STRING,\n" +" ts TIMESTAMP\n" +") WITH (\n" +" 'connector' = 'kafka',\n" +" 'topic' = 'routine_load_test2',\n" +" 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',\n" +" 'properties.group.id' = 'testGroup222',\n" +" 'format' = 'json',\n" +" 'scan.startup.mode' = 'earliest-offset'\n" +")";

进入flink sql client：

$FLINK_HOME/bin/sql-client.sh embedded -j /wyyt/software/flink-1.12.2-2.11/flink-1.12.2/lib/hudi-flink-bundle_2.1?-*.*.*.jar shell
---0，设置参数
set execution.result-mode=tableau;
SET table.exec.resource.default-parallelism = 4;

---1,创建kafka表

CREATE TABLE data_gen (
id STRING,
name STRING,
user_age STRING,
user_other STRING,
ts TIMESTAMP(3)
) WITH (
'connector' = 'kafka',
'topic' = 'routine_load_test2',
'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',
'properties.group.id' = 'testGroup3',
'format' = 'json',
'scan.startup.mode' = 'earliest-offset'
);

---2，创建hudi表

COW模式：

CREATE TABLE hudi_cow_data_gen(
id STRING,
name STRING,
user_age STRING,
user_other STRING,
ts TIMESTAMP(3),
PRIMARY KEY(id) NOT ENFORCED
)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen',
'table.type' = 'COPY_ON_WRITE',
'write.insert.drop.duplicates' = 'true'
);

MOR模式：

CREATE TABLE hudi_mor_data_gen(
id STRING,
name STRING,
user_age STRING,
user_other STRING,
ts TIMESTAMP(3),
PRIMARY KEY(id) NOT ENFORCED
)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen',
'table.type' = 'MERGE_ON_READ',
'read.streaming.enabled' = 'true',
'write.tasks'= '4',
'compaction.tasks'= '4',
'compaction.delta_seconds' = '5',
'compaction.delta_commits' = '1',
'read.streaming.check-interval' = '1'
);

--分区表
CREATE TABLE hudi_mor_data_gen(
id STRING,
name STRING,
user_age STRING,
user_other STRING,
ts TIMESTAMP(3),
sdt STRING,
PRIMARY KEY(id) NOT ENFORCED
)
partitioned by (sdt)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen',
'table.type' = 'MERGE_ON_READ',
'read.streaming.enabled' = 'true',
'write.tasks'= '4',
'compaction.tasks'= '4',
'compaction.delta_seconds' = '5',
'compaction.delta_commits' = '1',
'read.streaming.check-interval' = '1'
);

CREATE TABLE hudi_mor_data_gen2(
id STRING,
name STRING,
user_age STRING,
user_other STRING,
ts TIMESTAMP(3),
PRIMARY KEY(id) NOT ENFORCED
)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen2',
'table.type' = 'MERGE_ON_READ',
'read.streaming.enabled' = 'true',
'write.tasks'= '4',
'compaction.tasks'= '4',
'compaction.delta_seconds' = '180',
'compaction.delta_commits' = '3',
'read.streaming.check-interval' = '10'
);

---3，写入hudi

insert into hudi_cow_data_gen select * from data_gen;
insert into hudi_mor_data_gen select * from data_gen;

查询：
select count(*) from hudi_data_gen;

select count(*) from hudi_mor_data_gen;

---4, 创建hive表

进入hive client

hive

导入依赖：
ADD JAR /wyyt/software/flink-1.12.2-2.11/flink-1.12.2/hudi-hadoop-mr-bundle-0.9.0-SNAPSHOT.jar;

COW模式:

CREATE EXTERNAL TABLE `hive_cow_data_gen`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`id` string,
`name` string,
`user_age` string,
`user_other` string,
`ts` string)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.HoodieParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://bi-524:8020/tmp/default/hudi_assure_orders';

CREATE EXTERNAL TABLE `hive_assure_orders`(
`id` string,
`name` string,
`user_age` string,
`user_other` string
)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.HoodieParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://bi-524:8020/tmp/default/hudi_assure_orders';

MOR模式:

CREATE EXTERNAL TABLE `hive_mor_data_gen`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`id` INT,
`name` STRING,
`user_age` INT,
`user_other` STRING,
`ts` STRING)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen2';

CREATE EXTERNAL TABLE `hive_mor_data_gen`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`id` STRING,
`name` STRING,
`user_age` STRING,
`user_other` STRING,
)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://bi-524:8020/tmp/default/hudi_mor_data_gen2';

CREATE EXTERNAL TABLE `hudi_users_55`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`order_number` string,
`order_key` string,
`ts` bigint)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'hdfs://bi-524:8020/tmp/default/hudi_order_ods_test2';

========================================文档===========================================

一，数据湖？

数据湖（Data Lake)，湖仓一体（Data Lakehouse）俨然已经成为了大数据领域最为火热的流行词之一，各大主流公司都有在研究使用，也有相关的资料分享。

数据湖不是一个新鲜的东西，其实早就出来数据湖这个概念，初识数据湖肯定会有疑问：

数据湖是什么？

它到底解决了什么问题，拥有什么样新的特性呢？

它的现状是什么，还存在什么问题呢？

针对上面的问题，下面一一阐述：

数据湖是什么？

数据湖这样的一种数据存储结构，它可以存储任何类型的数据，包括像图片、文档这样的非结构化数据。数据湖通常更大，其存储成本也更为廉价。存储其中的数据不需要满足特定的schema，数据湖也不会尝试去将特定的schema施行其上。相反的是，数据的拥有者通常会在读取数据的时候解析schema（schema-on-read），当处理相应的数据时，将转换施加其上。

如果需要给数据湖下一个定义，可以定义为这样：数据湖是一个存储企业的各种各样原始数据的大型仓库，其中的数据可供存取、处理、分析及传输。

它到底解决了什么问题，拥有什么样新的特性呢？

数据湖中数据生命周期如图：

数据处理与分析既可按批量（batch）方式处理，也可以按近实时（near-real-time）方式处理。

数据湖与数据仓库的区别

数据湖	数据仓库
能处理所有类型的数据，如结构化数据，非结构化数据，半结构化数据等，数据的类型依赖于数据源系统的原始数据格式。	只能处理结构化数据进行处理，而且这些数据必须与数据仓库事先定义的模型吻合。
拥有足够强的计算能力用于处理和分析所有类型的数据，分析后的数据会被存储起来供用户使用。	处理结构化数据，将它们或者转化为多维数据，或者转换为报表，以满足后续的高级报表及数据分析需求。
数据湖通常包含更多的相关的信息，这些信息有很高概率会被访问，并且能够为企业挖掘新的运营需求。	数据仓库通常用于存储和维护长期数据，因此数据可以按需访问。

▲表2-1 数据湖与数据仓库的关键区别

数据湖与数据仓库的差别很明显。然而，在企业中两者的作用是互补的，不应认为数据湖的出现是为了取代数据仓库，毕竟两者的作用是截然不同的。

总结：数据湖的发展方向就是湖仓一体，是各种数据格式可以共存，满足近实时的批量一体数据处理。

它的现状是什么，还存在什么问题呢？
数据湖三剑客：

Iceberg

Hudi

Delta Lake

Iceberg 简介

iceberg优势？

优化数据入库流程：Iceberg 提供 ACID 事务能力，上游数据写入即可见，不影响当前数据处理任务，这大大简化了 ETL；Iceberg 提供了 upsert、merge into 能力，可以极大地缩小数据入库延迟；

支持更多的分析引擎：优秀的内核抽象使之不绑定特定的计算引擎，目前 Iceberg 支持的计算引擎有 Spark、Flink、Presto 以及 Hive；

统一数据存储和灵活的文件组织：提供了基于流式的增量计算模型和基于批处理的全量表计算模型。批处理和流任务可以使用相同的存储模型，数据不再孤立；Iceberg 支持隐藏分区和分区进化，方便业务进行数据分区策略更新。支持 Parquet、Avro 以及 ORC 等存储格式；

增量读取处理能力：Iceberg 支持通过流式方式读取增量数据，支持 Structed Streaming 以及 Flink table Source。

特点

模式演化支持添加，删除，更新或重命名，并且没有副作用；

隐藏分区可以防止导致错误提示或非常慢查询的用户错误；

分区布局演变可以随着数据量或查询模式的变化而更新表的布局；

快照控制可实现使用完全相同的表快照的可重复查询，或者使用户轻松检查更改；

版本回滚使用户可以通过将表重置为良好状态来快速纠正问题；

快速扫描数据无需使用分布式SQL引擎即可读取表或查找文件；

数据修剪优化使用表元数据使用分区和列级统计信息修剪数据文件；

兼容性好 ,可以存储在任意的云存储系统和HDFS中；

支持事务序列化隔离表更改是原子性的，读者永远不会看到部分更改或未提交的更改；

高并发高并发写入器使用乐观并发，即使写入冲突，也会重试以确保兼容更新成功。

总结：iceberg专业性目的性很强，在后续对象存储方面更具有优势，但是目前版本不支持修改删除，社区正在开发中，预计下一个版本。

Hudi简介

Hudi优势？

出现的更早，起初是跟spark深度绑定，目前支持Flink，支持修改删除功能。

了解hudi要知道的2个table.type （建表指定）

1. Copy On Write
Copy On Write 类型表每次写入都会生成一个新的持有 base file（对应写入的 instant time ）的 FileSlice。

用户在 snapshot 读取的时候会扫描所有最新的 FileSlice 下的 base file。

2. Merge On Read

Merge On Read 表的写入行为，依据 index 的不同会有细微的差别：

•对于 BloomFilter 这种无法对 log file 生成 index 的索引方案，对于 INSERT 消息仍然会写 base file （parquet format），只有 UPDATE 消息会 append log 文件（因为 base file 已经记录了该 UPDATE 消息的 FileGroup ID）。•对于可以对 log file 生成 index 的索引方案，例如 Flink writer 中基于 state 的索引，每次写入都是 log format，并且会不断追加和 roll over。

Merge On Read 表的读在 READ OPTIMIZED 模式下，只会读最近的经过 compaction 的 commit。

为什么Hudi对于大规模和近实时应用很重要

首先我们讲一讲Flink+Hudi 与 hive的对比说明：

Hudi解决了以下限制：

HDFS的可伸缩性限制，支持多种数据格式

解决了小文件问题

支持对现有数据的更新和删除

近实时分析：相对于秒级存储 (Druid, OpenTSDB) 更节省资源

提供分钟级别时效性，支撑更高效的查询

支撑更高效的查询: Hudi 作为 lib，非常轻量

增量导出 : 替代部分 Kafka 的场景

可以实现批流一体，一套架构，减少资源的使用，特别适合中小公司。

二，应用实践

Hudi集成Flink（以Hudi为例）

Iceberg 深度集成 Flink（略）

前面讲到，我们的绝大部分任务都是 Flink 任务，包括批处理任务和流处理任务，目前这三个数据湖框架，Iceberg 是集成 Flink 做的最完善的，如果采用 Iceberg 替代 Hive 之后，迁移的成本非常小，对用户几乎是无感知的，

比如我们原来的 SQL 是这样的：

INSERT INTO hive_catalog.db.hive_table SELECT * FROM kafka_table

迁移到 Iceberg 以后，只需要修改 catalog 就行。

INSERT INTO iceberg_catalog.db.iIcebergceberg_table SELECT * FROM kafka_table

Presto 查询也是和这个类似，只需要修改 catalog 就行了。

Iceberg 的设计架构使得查询更快

在 Iceberg 的设计架构中，manifest 文件存储了分区相关信息、data files 的相关统计信息（max/min）等，去查询一些大的分区的数据，就可以直接定位到所要的数据，而不是像 Hive 一样去 list 整个 HDFS 文件夹，时间复杂度从 O(n) 降到了 O(1)，使得一些大的查询速度有了明显的提升，在 Iceberg PMC Chair Ryan Blue 的演讲中，我们看到命中 filter 的任务执行时间从 61.5 小时降到了 22 分钟。

三，快速上手，案例演示

Iceberg（略）

1，Zeppelin案例演示：

http://bi-525:18080/#/notebook/2G5A772Y6

2，iceberg集成hive演示

1）下载jar
地址：https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-hive-runtime/0.11.0/

2）2

3）3

Hudi

页面演示：

http://bi-525:18080/#/notebook/2G3HP31EP

zeppelin案例演示

这里将 table optionread.streaming.enabled 设置为 true，表明通过 streaming 的方式读取表数据，opitonread.streaming.check-interval 指定了 source 监控新的 commits 的间隔为 4s，optiontable.type 设置表类型为 MERGE_ON_READ，目前只有 MERGE_ON_READ 表支持 streaming 读。

五，最终期望

期望实现架构：

行业参考：

T3 出行是如何构建湖仓一体的：

架构：

Hudi 插件化的架构

Hudi 存储模式与视图：

准实时分析流程：

六，集群环境整理，记录点

Flink on zeppelin

1）Zeppelin flink 参数配置：

Flink sql 读写hive

1）Jar包放入Flink lib下：

flink-connector-hive_2.11-1.11.2.jar

hive-exec-2.1.0.jar

2）可能需要导入hadoop包
flink-hadoop-compatibility_2.11-1.11.2.jar

flink-shaded-hadoop-2-uber-2.6.0-cdh5.16.2-9.0.jar（这要自己编译）
flink-shaded-hadoop-2-uber-2.7.5-9.0.jar（实际测试使用这个）

3) 配置环境变量

实际配置为：

export HADOOP_CLASSPATH=`hadoop classpath`

export HADOOP_CONF_DIR=/etc/hadoop/conf.cloudera.hdfs

export HADOOP_HOME=/wyyt/software/cloudera/parcels/CDH/lib/hadoop

export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH

export HIVE_CONF_DIR=/etc/hive/conf

export HIVE_HOME=/wyyt/software/cloudera/parcels/CDH/lib/hive

export PATH=$HIVE_HOME/bin:$HIVE_HOME/sbin:$PATH

4）还可以参考自己的文章
https://blog.csdn.net/qq_31866793/article/details/107466561

阿里文章：

https://developer.aliyun.com/article/761469

Flink 读写iceberg

Flink 1.12.2版本：