hudi的操作记录备份以及文档

逃离我推掉我的手 2023-01-21 15:26 69阅读 0赞

# 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），当处理相应的数据时，将转换施加其上。

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

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

**数据湖与数据仓库的区别**

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

▲表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。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

**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。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

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 的设计架构使得查询更快

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

在 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][http_bi-525_18080_notebook_2G5A772Y6]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16][]

### 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][http_bi-525_18080_notebook_2G3HP31EP]

1.  zeppelin案例演示

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17][]

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

# 五，最终期望 #

### 期望实现架构： ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 18][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 19][]

### 行业参考： ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 20][]

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

**架构：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 21][]

**Hudi** **插件化的架构**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 22][]

**Hudi** **存储模式与视图：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 23][]

**准实时分析流程：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 24][]

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

*  Flink on zeppelin

1）Zeppelin flink 参数配置：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 25][]

*  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) 配置环境变量

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 26][]

实际配置为：

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 27][]

4）还可以参考自己的文章  
       [https://blog.csdn.net/qq\_31866793/article/details/107466561][https_blog.csdn.net_qq_31866793_article_details_107466561]

阿里文章：

[https://developer.aliyun.com/article/761469][https_developer.aliyun.com_article_761469]

*  Flink 读写iceberg

Flink 1.12.2版本：

[http://bi-525:18080/\#/notebook/2G5A772Y6][http_bi-525_18080_notebook_2G5A772Y6]

>        %flink.conf
> 
>        flink.execution.packages org.apache.flink:flink-connector-kafka\_2.11:1.12.2
> 
>        flink.execution.jars /wyyt/software/flink-1.12.2-2.11/flink-1.12.2/lib/iceberg-flink-runtime-dae6c49.jar
> 
>        execution.checkpointing.interval 10000
> 
>        execution.checkpointing.externalized-checkpoint-retention RETAIN\_ON\_CANCELLATION
> 
>        state.backend filesystem
> 
>        state.checkpoints.dir hdfs:///flink/checkpoints

*  Flink 读写iceberg 整合hive  
    Flink 1.12.2版本

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 28][]

*  Flink读写hudi 整合hive

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 29][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 30][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221020/de87225da60440b19a0a5ad9f10e017b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221020/28c39f77acfb4eb19befb06f8e28c1fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221020/16e7e4c7a00b4eb7b8cff100868c9772.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221020/14fc126490b54cc397ffee6a501fbbc9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221020/8345198e55914fc7b6dc9f5c40f896f8.png
[http_bi-525_18080_notebook_2G5A772Y6]: http://bi-525:18080/#/notebook/2G5A772Y6
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221020/5c12860d31094c9bb740337d97a70f9d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6]: /images/20221020/df888170631b448e87aa43ca894026db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7]: /images/20221020/de3d480dd77d45b19236ea314232e767.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8]: /images/20221020/ffbe46c55cff457d8c6bea0532769a96.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9]: /images/20221020/11603d0e320b4b7285b3a75d565e44e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10]: /images/20221020/b2b65b2910984ae8923076f5ea0c13b5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11]: /images/20221020/8fc34e8207664987bf7597f9513cad4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12]: /images/20221020/f6249e37fad449c5a3beab059b43d309.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13]: /images/20221020/b5addbe34090490ebc89eeb3c0f2c8e3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14]: /images/20221020/c18f739b5edd476f9a3eb84962189f17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15]: /images/20221020/7802bc76e42c4e38aea0d886cc09b81b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16]: /images/20221020/b4605fc0acb343ed895b54082a2537ac.png
[http_bi-525_18080_notebook_2G3HP31EP]: http://bi-525:18080/#/notebook/2G3HP31EP
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17]: /images/20221020/7a924bc49857483ebc916701890eb44a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 18]: /images/20221020/0a5b13185a754e8b8b5398f7c6645de5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 19]: /images/20221020/dbea9193dfad4306bd9ffd80e9363dcb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 20]: /images/20221020/9261fb55a2924c06b7a8982487a02218.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 21]: /images/20221020/4e3431c35b564f99aa6cfb586182ddd6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 22]: /images/20221020/cf5c1c98e50e42f18826b3d3fcc7a733.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 23]: /images/20221020/7821fd8888dc4cf09705fea3086b0d56.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 24]: /images/20221020/a46df98a570a465698a20e624eee6f5e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 25]: /images/20221020/10f32597aadd432b9e67ffc2692aa45a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 26]: /images/20221020/8440d5362faa4771a75d5c0c73389c10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 27]: /images/20221020/8be907a91e814685848c0fc6e77c85c4.png
[https_blog.csdn.net_qq_31866793_article_details_107466561]: https://blog.csdn.net/qq_31866793/article/details/107466561
[https_developer.aliyun.com_article_761469]: https://developer.aliyun.com/article/761469
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 28]: /images/20221020/e9b9ba2fde984f5d845abc46d26448b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 29]: /images/20221020/d55897fa4fcb4aeeb7cbcbf34de4da42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 30]: /images/20221020/0c37e8a7953a47b396addc2db993befd.png