Hudi 数据湖技术之集成Flink

冷不防 2024-03-22 09:08 208阅读 0赞

#### 目录 ####

*  1 安装Flink
 *  2 快速入门
 *   *  2.1 集成Flink概述
     *  2.2 环境准备
     *  2.3 创建表
     *  2.4 插入数据
     *  2.5 查询数据
     *  2.6 更新数据
 *  3 Streaming query
 *   *  3.1 创建表
     *  3.2 查询数据
     *  3.3 插入数据
 *  4 Flink SQL Writer
 *   *  4.1 Flink SQL集成Kafka
     *  4.2 Flink SQL写入Hudi
     *   *  4.2.1 创建Maven Module
         *  4.2.2 消费Kafka数据
         *  4.2.3 保存数据至Hudi
         *  4.2.4 加载Hudi表数据
     *  4.3 Flink SQL Client 写入Hudi
     *   *  4.3.1 集成环境
         *  4.3.2 执行SQL
 *  5 Hudi CDC
 *   *  5.1 CDC 数据入湖
     *  5.2 Flink CDC Hudi
     *   *  5.2.1 业务需求
         *  5.2.2 创建 MySQL 表
         *  5.2.3 创建 CDC 表
         *  5.2.4 创建视图
         *  5.2.5 创建 Hudi 表
         *  5.2.6 数据写入Hudi表
         *  5.2.7 Hive 表查询
     *  5.3 Hudi Client操作Hudi表

--------------------

## 1 安装Flink ##

![在这里插入图片描述][1eee962b1eaa4b19a81ff16feb3aff68.png]

使用Flink 1.12版本，部署Flink Standalone集群模式，启动服务，步骤如下：  
step1、下载安装包

> https://archive.apache.org/dist/flink/flink-1.12.2/

step2、上传软件包

> flink-1.12.2-bin-scala\_2.12.tgz 到node1的指定目录

step3、解压

> tar -zxvf flink-1.12.2-bin-scala\_2.12.tgz -C /export/server/ chown -R  
> root:root /export/server/flink-1.12.2/

step4、创建软连接

> ln -s flink-1.12.2 flink

step5、添加hadoop依赖jar包

> cd /export/server/flink/lib  
> 使用rz上传jar包：flink-shaded-hadoop-2-uber-2.7.5-10.0.jar

![在这里插入图片描述][545f480c34384e04bc2c5c59b08184c2.png]

step6、启动HDFS集群

> hadoop-daemon.sh start namenode  
> hadoop-daemon.sh start datanode

step7、启动Flink本地集群

> /export/server/flink/bin/start-cluster.sh  
> 使用jps可以查看到下面两个进程

![在这里插入图片描述][325e194e817a40f08eb69deda1ffd444.png]

停止Flink

> /export/server/flink/bin/stop-cluster.sh

step8、访问Flink的Web UI  
网址：node1:8081/\#/overview  
![在这里插入图片描述][884cd38d98d3484790cd8aebeba881d9.png]

step9、执行官方示例  
读取文本文件数据，进行词频统计WordCount，将结果打印控制台。

> /export/server/flink/bin/flink run  
> /export/server/flink/examples/batch/WordCount.jar

## 2 快速入门 ##

基于Flink操作Hudi表数据，进行查询分析，软件版本说明如下：  
![在这里插入图片描述][68accaf67cab401789d95e5940ee6a78.png]

### 2.1 集成Flink概述 ###

Flink集成Hudi时，本质将集成jar包：hudi-flink-bundle\_2.12-0.9.0.jar，放入Flink 应用CLASSPATH下即可。Flink SQLConnector支持Hudi作为Source和Sink时，两种方式将jar包放入CLASSPATH路径：  
● 方式一：运行Flink SQL Client命令行时，通过参数【-j xx.jar】指定jar包  
![在这里插入图片描述][41b0ada4afd042a2bb72e80cd10c9536.png]

● 方式二：将jar包直接放入Flink软件安装包lib目录下【$FLINK\_HOME/lib】

![在这里插入图片描述][07a279397d704dc885f506cf43bdf7f5.png]

接下来使用Flink SQL Client提供SQL命令行与Hudi集成，需要启动Flink Standalone集群，其中需要修改配置文件【$FLINK\_HOME/conf/flink-conf.yaml】，TaskManager分配Slots数目为4。  
![在这里插入图片描述][c3589734b25a4b4b892f6892ab157468.png]

### 2.2 环境准备 ###

首先启动各个框架服务，然后编写DDL语句创建表，最后DML语句进行插入数据和查询分析。按照如下步骤启动环境，总共分为三步：  
● 第一步、启动HDFS集群

> \[root@node1 ~\]\# hadoop-daemon.sh start namenode  
> \[root@node1 ~\]\# hadoop-daemon.sh start datanode

● 第二步、启动Flink 集群  
由于Flink需要连接HDFS文件系统，所以先设置HADOOP\_CLASSPATH变量，再启动Standalone集群服务。

> \[root@node1 ~\]\# export HADOOP\_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
> 
> \[root@node1 ~\]\# /export/server/flink/bin/start-cluster.sh

● 第三步、启动Flink SQL Cli命令行

> \[root@node1 ~\]\# /export/server/flink/bin/sql-client.sh embedded shell

采用指定参数【-j xx.jar】方式加载hudi-flink集成包，命令如下。

> \[root@node1 ~\]\# /export/server/flink/bin/sql-client.sh embedded -j  
> /root/hudi-flink-bundle\_2.11-0.9.0.jar shell

在SQL Cli设置分析结果展示模式为：set execution.result-mode=tableau;。  
![在这里插入图片描述][ff4af9957c3b42cebbf4f31dc9433111.png]

### 2.3 创建表 ###

创建表：t1，数据存储到Hudi表中，底层HDFS存储，表的类型：MOR，语句如下：

CREATE TABLE t1(
      uuid VARCHAR(20), 
      name VARCHAR(10),
      age INT,
      ts TIMESTAMP(3),
      `partition` VARCHAR(20)
    )
    PARTITIONED BY (`partition`)
    WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://node1.oldlu.cn:8020/ehualu/hudi-warehouse/hudi-t1',
      'write.tasks' = '1',
      'compaction.tasks' = '1', 
      'table.type' = 'MERGE_ON_READ'
    );

在Flink SQL CLI命令行执行DDL语句，截图如下所示：  
![在这里插入图片描述][6a259a51a212418abd540dc08c28faca.png]

查看表及结构，命令如下：  
![在这里插入图片描述][89ebd7441d624718904c24c5a188c951.png]  
接下来，编写INSERT语句，向Hudi表中插入数据。

### 2.4 插入数据 ###

向上述创建表：t1中插入数据，其中t1表为分区表，字段名称：partition，插入数据时字段值有：【part1、part2、part3和part4】，语句如下：

INSERT INTO t1 VALUES
    ('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1');
    
    INSERT INTO t1 VALUES
    ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),
    ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),
    ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),
    ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),
    ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),
    ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),
    ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');

在Flink SQL CLI中执行截图如下：  
![在这里插入图片描述][dc3e3fccc3094a4eb2b2e03fa57ea4e5.png]

日志信息可知，将SQL语句提交至Flink Standalone集群执行，并且insert语句执行成功。  
![在这里插入图片描述][b84d6134fc51415881ad0c136632f944.png]  
查询HDFS上数据存储目录：  
![在这里插入图片描述][406f56c1a880446390be3d60b87e9c1a.png]

### 2.5 查询数据 ###

数据通过Flink SQL CLi插入Hudi表后，编写SQL语句查询数据，语句如下：

> select \* from t1;

与插入数据一样，向Standalone集群提交SQL，生成Job查询数据。  
![在这里插入图片描述][f37746ea808946e79126869a3ff2143f.png]

通过在 WHERE 子句中添加 partition 路径来裁剪 partition，如下所示：

> select \* from t1 where `partition` = ‘par1’ ;  
> ![在这里插入图片描述][0af16cf4db1d4508b8651f0ce46839d6.png]

### 2.6 更新数据 ###

将id1的数据age由23变为了27，执行SQL语句如下：

> insert into t1 values (‘id1’,‘Danny’,27,TIMESTAMP ‘1970-01-01  
> 00:00:01’,‘par1’);

再次查询表的数据，结果如下：  
![在这里插入图片描述][73b2b4b1beb049da868f50ac1e9e7b79.png]

插入Flink Standalone监控页面8081，可以看到执行3个job。  
![在这里插入图片描述][77fcfc6ef4bf489fa2fb96ca44fcf38f.png]

## 3 Streaming query ##

Flink插入Hudi表数据时，支持以流的方式加载数据，增量查询分析。

### 3.1 创建表 ###

首先创建表：t2，设置相关属性，以流的方式查询读取，映射到前面表：t1，语句如下。

CREATE TABLE t2(
      uuid VARCHAR(20), 
      name VARCHAR(10),
      age INT,
      ts TIMESTAMP(3),
      `partition` VARCHAR(20)
    )
    PARTITIONED BY (`partition`)
    WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://node1.oldlu.cn:8020/hudi-warehouse/hudi-t1',
      'table.type' = 'MERGE_ON_READ',
      'read.tasks' = '1', 
      'read.streaming.enabled' = 'true',
      'read.streaming.start-commit' = '20210316134557',
      'read.streaming.check-interval' = '4' 
    );

核心参数选项说明：  
● read.streaming.enabled 设置为 true，表明通过 streaming 的方式读取表数据；  
● read.streaming.check-interval 指定了 source 监控新的 commits 的间隔为 4s；  
● table.type 设置表类型为 MERGE\_ON\_READ；  
![在这里插入图片描述][60cb1771ead347708d865f095205d27a.png]

接下来编写SQL插入数据，流式方式插入表：t2数据。

### 3.2 查询数据 ###

创建表：t2 以后，此时表的数据就是前面批Batch模式写入的数据。

> select \* from t2 ;

![在这里插入图片描述][9176e21665bc4f9b9d518c4f4d44240a.png]

插入显示表中所有数据，光标在一直闪动，每隔4秒，再依据commit timestamp增量查询。

### 3.3 插入数据 ###

重新开启Terminal启动Flink SQL CLI，重新创建表：t1，采用批Batch模式插入1条数据。

CREATE TABLE t1(
      uuid VARCHAR(20), 
      name VARCHAR(10),
      age INT,
      ts TIMESTAMP(3),
      `partition` VARCHAR(20)
    )
    PARTITIONED BY (`partition`)
    WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://node1.oldlu.cn:8020/hudi-warehouse/hudi-t1',
      'write.tasks' = '1',
      'compaction.tasks' = '1', 
      'table.type' = 'MERGE_ON_READ'
    );
    
    insert into t1 values ('id9','test',27,TIMESTAMP '1970-01-01 00:00:01','par5');

几秒后在流表中可以读取到一条新增的数据（前面插入的一条数据）。  
![在这里插入图片描述][914a3f2b49174aa9abbb790fe8c5e076.png]

通过一些简单的演示，发现 HUDI Flink 的集成已经相对完善，读写数据均已覆盖。

## 4 Flink SQL Writer ##

在hudi-flink模块中提供Flink SQL Connector连接器，支持从Hudi表读写数据。  
![在这里插入图片描述][1d2c41ef02f7464db0261fac0a3344c2.png]

文档：https://hudi.apache.org/docs/writing\_data\#flink-sql-writer

### 4.1 Flink SQL集成Kafka ###

首先配置Flink SQL 集成Kafka，实时消费Kafka Topic数据，具体操作如下步骤：  
![在这里插入图片描述][75ab8e15c9b24eda91107e9f2c27d690.png]

● 第一步、创建Topic  
启动Zookeeper和Kafka服务组件，案例演示FlinkSQL与Kafka集成，实时加载数据。使用KafkaTool工具，连接启动Kafka服务，创建topic：flink-topic。  
![在这里插入图片描述][41ade2f3297c48a1bf902fbe3cbdc526.png]

可以使用命令行创建Topic，具体命令如下：  
– 创建topic：flink-topic

> kafka-topics.sh --create --bootstrap-server node1.oldlu.cn:9092  
> –replication-factor 1 --partitions 1 --topic flink-topic

启动Flink Standalone集群服务，运行flink-sql命令行，创建表映射到Kafka中。

■第二步、启动HDFS集群

> \[root@node1 ~\]\# hadoop-daemon.sh start namenode \[root@node1 ~\]\#  
> hadoop-daemon.sh start datanode

■第三步、启动Flink 集群  
由于Flink需要连接HDFS文件系统，所以先设置HADOOP\_CLASSPATH变量，再启动Standalone集群服务。

> \[root@node1 ~\]\# export HADOOP\_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath` \[root@node1 ~\]\# /export/server/flink/bin/start-cluster.sh

■第四步、启动Flink SQL Cli命令行  
采用指定参数【-j xx.jar】方式加载hudi-flink集成包，命令如下。

> \[root@node1 ~\]\# cd /export/server/flink \[root@node1 ~\]\#  
> bin/sql-client.sh embedded -j  
> /root/flink-sql-connector-kafka\_2.11-1.12.0.jar shell

在SQL Cli设置分析结果展示模式为：tableau。  
![在这里插入图片描述][de7a3e2eb4ca423d9c143b74dacf346d.png]

■第五步、创建表，映射到Kafka Topic  
其中Kafka Topic中数据是CSV文件格式，有三个字段：user\_id、item\_id、behavior，从Kafka消费数据时，设置从最新偏移量开始，创建表语句如下：set execution.result-mode=tableau;

CREATE TABLE tbl_kafka (
      `user_id` BIGINT,
      `item_id` BIGINT,
      `behavior` STRING
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'flink-topic',
      'properties.bootstrap.servers' = 'node1.oldlu.cn:9092',
      'properties.group.id' = 'test-group-10001',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'csv'
    );

执行命令后，查看表，截图如下：  
![在这里插入图片描述][913a15cf9d044ea1a80be611a751ea8c.png]

■第六步、实时向Topic发送数据，并在FlinkSQL查询  
首先，在FlinkSQL页面，执行SELECT查询语句，截图如下：  
![在这里插入图片描述][5c66f362c5e34fa29f62a49e787593c7.png]

其次，通过Kafka Console Producer向Topic发送数据，命令和数据如下：  
– 生产者发送数据  
kafka-console-producer.sh --broker-list node1.oldlu.cn:9092 --topic flink-topic

/*
    1001,90001,click
    1001,90001,browser
    1001,90001,click
    1002,90002,click
    1002,90003,click
    1003,90001,order
    1004,90001,order
    */

插入数据，观察FlinkSQL界面，可以发现数据实时查询处理，截图如下所示：  
![在这里插入图片描述][9ae9108a3cb24a69b5e2e941cd90efe4.png]

至此FlinkSQL集成Kafka，采用表的方式关联Topic数据，接下来编写Flink SQL 程序实时将Kafka数据同步到Hudi表中。

### 4.2 Flink SQL写入Hudi ###

将上述编写StructuredStreaming流式程序改为Flink SQL程序：实时从Kafka消费Topic数据，解析转换后，存储至Hudi表中，示意图如下所示。  
![在这里插入图片描述][6c9058d04f734d25a3c55baa73374455.png]

#### 4.2.1 创建Maven Module ####

![在这里插入图片描述][a60f851aecd2402c89dcb27a937ed084.png]

创建Maven Module模块，添加依赖，此处Flink：1.12.2和Hudi：0.9.0版本。

<repositories>
        <repository>
            <id>nexus-aliyun</id>
            <name>Nexus aliyun</name>
            <url>http://maven.aliyun.com/nexus/content/groups/public</url>
        </repository>
        <repository>
            <id>central_maven</id>
            <name>central maven</name>
            <url>https://repo1.maven.org/maven2</url>
        </repository>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
        <repository>
            <id>apache.snapshots</id>
            <name>Apache Development Snapshot Repository</name>
            <url>https://repository.apache.org/content/repositories/snapshots/</url>
            <releases>
                <enabled>false</enabled>
            </releases>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </repository>
    </repositories>
    
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
        <java.version>1.8</java.version>
        <scala.binary.version>2.12</scala.binary.version>
        <flink.version>1.12.2</flink.version>
        <hadoop.version>2.7.3</hadoop.version>
        <mysql.version>8.0.16</mysql.version>
    </properties>
    
    <dependencies>
        
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.hudi</groupId>
            <artifactId>hudi-flink-bundle_${scala.binary.version}</artifactId>
            <version>0.9.0</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-hadoop-2-uber</artifactId>
            <version>2.7.5-10.0</version>
        </dependency>
    
        
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>${mysql.version}</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.68</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>
    
    </dependencies>
    
    <build>
        <sourceDirectory>src/main/java</sourceDirectory>
        <testSourceDirectory>src/test/java</testSourceDirectory>
        <plugins>
            
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.5.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.18.1</version>
                <configuration>
                    <useFile>false</useFile>
                    <disableXmlReport>true</disableXmlReport>
                    <includes>
                        <include>**/*Test.*</include>
                        <include>**/*Suite.*</include>
                    </includes>
                </configuration>
            </plugin>
            
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
    
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

#### 4.2.2 消费Kafka数据 ####

创建类：FlinkSQLKafakDemo，基于Flink Table API，从Kafka消费数据，提取字段值（方便后续存储Hudi表中）。

package cn.oldlu.hudi;
    
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.TableEnvironment;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    /**
     * 基于Flink SQL Connector实现：实时消费Topic中数据，转换处理后，实时存储Hudi表中
     */
    public class FlinkSQLKafakDemo {
        
    
       public static void main(String[] args) {
        
    
          // 1-获取表执行环境
          EnvironmentSettings settings = EnvironmentSettings
             .newInstance()
             .inStreamingMode()
             .build();
          TableEnvironment tableEnv = TableEnvironment.create(settings) ;
    
          // 2-创建输入表, TODO: 从Kafka消费数据
          tableEnv.executeSql(
             "CREATE TABLE order_kafka_source (\n" +
                "  orderId STRING,\n" +
                "  userId STRING,\n" +
                "  orderTime STRING,\n" +
                "  ip STRING,\n" +
                "  orderMoney DOUBLE,\n" +
                "  orderStatus INT\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'order-topic',\n" +
                "  'properties.bootstrap.servers' = 'node1.oldlu.cn:9092',\n" +
                "  'properties.group.id' = 'gid-1001',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json',\n" +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true'\n" +
                ")"
          );
    
          // 3-数据转换：提取订单时间中订单日期，作为Hudi表分区字段值
          Table etlTable = tableEnv
             .from("order_kafka_source")
             .addColumns(
                $("orderTime").substring(0, 10).as("partition_day")
             )
             .addColumns(
                $("orderId").substring(0, 17).as("ts")
             );
          tableEnv.createTemporaryView("view_order", etlTable);
    
          // 4-查询数据
          tableEnv.executeSql("SELECT * FROM view_order").print();
       }
    
    }

运行流式应用程序和模拟数据程序，查看控制台。  
![在这里插入图片描述][48d12fd944a047918162ac794fca909c.png]

#### 4.2.3 保存数据至Hudi ####

编写创建表DDL语句，映射到Hudi表中，指定相关属性：主键字段、表类型等等。

CREATE TABLE order_hudi_sink (
      orderId STRING PRIMARY KEY NOT ENFORCED,
      userId STRING,
      orderTime STRING,
      ip STRING,
      orderMoney DOUBLE,
      orderStatus INT,
      ts STRING,
      partition_day STRING
    )
    PARTITIONED BY (partition_day)
    WITH (
        'connector' = 'hudi',
        'path' = 'file:///D:/flink_hudi_order',
        'table.type' = 'MERGE_ON_READ',
        'write.operation' = 'upsert',
        'hoodie.datasource.write.recordkey.field'= 'orderId',
        'write.precombine.field' = 'ts',
        'write.tasks'= '1'
    );

将Hudi表数据保存在本地文件系统LocalFS目录中，此外，向Hudi表写入数据时，采用INSERT INTO插入方式写入数据，具体DDL语句如下：  
– 子查询插入INSERT … SELECT …

INSERT INTO order_hudi_sink
    SELECT
        orderId, userId, orderTime, ip, orderMoney, orderStatus,
        substring(orderId, 0, 17) AS ts, substring(orderTime, 0, 10) AS partition_day
    FROM order_kafka_source ;
    12345

创建类：FlinkSQLHudiDemo，编写代码：从Kafka消费数据，转换后，保存到Hudi表。

package cn.oldlu.hudi;
    
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    import static org.apache.flink.table.api.Expressions.$;
    
    /**
     * 基于Flink SQL Connector实现：实时消费Topic中数据，转换处理后，实时存储Hudi表中
     */
    public class FlinkSQLHudiDemo {
        
    
       public static void main(String[] args) {
        
    
          System.setProperty("HADOOP_USER_NAME","root");
    
          // 1-获取表执行环境
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          env.setParallelism(1);
          env.enableCheckpointing(5000);
          EnvironmentSettings settings = EnvironmentSettings
             .newInstance()
             .inStreamingMode()
             .build();
          StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings) ;
    
          // 2-创建输入表, TODO: 从Kafka消费数据
          tableEnv.executeSql(
             "CREATE TABLE order_kafka_source (\n" +
                "  orderId STRING,\n" +
                "  userId STRING,\n" +
                "  orderTime STRING,\n" +
                "  ip STRING,\n" +
                "  orderMoney DOUBLE,\n" +
                "  orderStatus INT\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'order-topic',\n" +
                "  'properties.bootstrap.servers' = 'node1.oldlu.cn:9092',\n" +
                "  'properties.group.id' = 'gid-1001',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json',\n" +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true'\n" +
                ")"
          );
    
          // 3-数据转换：提取订单时间中订单日期，作为Hudi表分区字段值
          Table etlTable = tableEnv
             .from("order_kafka_source")
             .addColumns(
                $("orderId").substring(0, 17).as("ts")
             )
             .addColumns(
                $("orderTime").substring(0, 10).as("partition_day")
             );
          tableEnv.createTemporaryView("view_order", etlTable);
    
          // 4-定义输出表，TODO：数据保存到Hudi表中
          tableEnv.executeSql(
             "CREATE TABLE order_hudi_sink (\n" +
                "  orderId STRING PRIMARY KEY NOT ENFORCED,\n" +
                "  userId STRING,\n" +
                "  orderTime STRING,\n" +
                "  ip STRING,\n" +
                "  orderMoney DOUBLE,\n" +
                "  orderStatus INT,\n" +
                "  ts STRING,\n" +
                "  partition_day STRING\n" +
                ")\n" +
                "PARTITIONED BY (partition_day) \n" +
                "WITH (\n" +
                "  'connector' = 'hudi',\n" +
                "  'path' = 'file:///D:/flink_hudi_order',\n" +
                "  'table.type' = 'MERGE_ON_READ',\n" +
                "  'write.operation' = 'upsert',\n" +
                "  'hoodie.datasource.write.recordkey.field' = 'orderId'," +
                "  'write.precombine.field' = 'ts'" +
                "  'write.tasks'= '1'" +
                ")"
          );
    
          // 5-通过子查询方式，将数据写入输出表
          tableEnv.executeSql(
             "INSERT INTO order_hudi_sink\n" +
                "SELECT\n" +
                "  orderId, userId, orderTime, ip, orderMoney, orderStatus, ts, partition_day\n" +
                "FROM view_order"
          );
    
       }
    
    }

运行上述编写流式程序，查看本地文件系统目录，保存Hudi表数据结构信息：  
![在这里插入图片描述][bef411c891b840d6b621fa9ac211e282.png]

#### 4.2.4 加载Hudi表数据 ####

创建类：FlinkSQLReadDemo，加载Hudi表中数据，采用流式方式读取，同样创建表，映射关联到Hudi表数据存储目录中，创建表DDL语句如下：

CREATE TABLE order_hudi(
      orderId STRING PRIMARY KEY NOT ENFORCED,
      userId STRING,
      orderTime STRING,
      ip STRING,
      orderMoney DOUBLE,
      orderStatus INT,
      ts STRING,
      partition_day STRING
    )
    PARTITIONED BY (partition_day)
    WITH (
        'connector' = 'hudi',
        'path' = 'file:///D:/flink_hudi_order',
        'table.type' = 'MERGE_ON_READ',
        'read.streaming.enabled' = 'true',
        'read.streaming.check-interval' = '4'
    );

完整Flink SQL流式程序代码如下：

package cn.oldlu.hudi;
    
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.TableEnvironment;
    
    /**
     * 基于Flink SQL Connector实现：从Hudi表中加载数据，编写SQL查询
     */
    public class FlinkSQLReadDemo {
        
    
       public static void main(String[] args) {
        
          System.setProperty("HADOOP_USER_NAME","root");
    
          // 1-获取表执行环境
          EnvironmentSettings settings = EnvironmentSettings
             .newInstance()
             .inStreamingMode()
             .build();
          TableEnvironment tableEnv = TableEnvironment.create(settings) ;
    
          // 2-创建输入表, TODO: 加载Hudi表查询数据
          tableEnv.executeSql(
             "CREATE TABLE order_hudi(\n" +
                "  orderId STRING PRIMARY KEY NOT ENFORCED,\n" +
                "  userId STRING,\n" +
                "  orderTime STRING,\n" +
                "  ip STRING,\n" +
                "  orderMoney DOUBLE,\n" +
                "  orderStatus INT,\n" +
                "  ts STRING,\n" +
                "  partition_day STRING\n" +
                ")\n" +
                "PARTITIONED BY (partition_day)\n" +
                "WITH (\n" +
                "  'connector' = 'hudi',\n" +
                "  'path' = 'file:///D:/flink_hudi_order',\n" +
                "  'table.type' = 'MERGE_ON_READ',\n" +
                "  'read.streaming.enabled' = 'true',\n" +
                "  'read.streaming.check-interval' = '4'\n" +
                ")"
          );
    
          // 3-通过子查询方式，将数据写入输出表
          tableEnv.executeSql(
             "SELECT \n" +
                "  orderId, userId, orderTime, ip, orderMoney, orderStatus, ts ,partition_day \n" +
                "FROM order_hudi"
          ).print();
    
       }
    
    }

运行流式程序，加载Hudi表数据，结果如下所示：  
![在这里插入图片描述][f9b58abdd31f4bcd9106d0c0c0ad0986.png]

### 4.3 Flink SQL Client 写入Hudi ###

启动Flink Standalone集群，运行SQL Client命令行客户端，执行DDL和DML语句，操作数据。

#### 4.3.1 集成环境 ####

■配置Flink 集群  
修改$FLINK\_HOME/conf/flink-conf.yaml文件

jobmanager.rpc.address: node1.oldlu.cn
    jobmanager.memory.process.size: 1024m
    taskmanager.memory.process.size: 2048m
    taskmanager.numberOfTaskSlots: 4
    
    classloader.check-leaked-classloader: false
    classloader.resolve-order: parent-first
    
    execution.checkpointing.interval: 3000
    state.backend: rocksdb
    state.checkpoints.dir: hdfs://node1.oldlu.cn:8020/flink/flink-checkpoints
    state.savepoints.dir: hdfs://node1.oldlu.cn:8020/flink/flink-savepoints
    state.backend.incremental: true

● 将Hudi与Flink集成jar包及其他相关jar包，放置到$FLINK\_HOME/lib目录  
![在这里插入图片描述][caad7bef601348f7961badb51955c9d3.png]

● 启动Standalone集群

> export HADOOP\_CLASSPATH=`/export/server/hadoop/bin/hadoop classpath`  
> /export/server/flink/bin/start-cluster.sh

● 启动SQL Client，最好再次指定Hudi集成jar包

> /export/server/flink/bin/sql-client.sh embedded -j  
> /export/server/flink/lib/hudi-flink-bundle\_2.12-0.9.0.jar shell

● 设置属性

> set execution.result-mode=tableau; set  
> execution.checkpointing.interval=3sec;

![在这里插入图片描述][75615a0034e447ddaae64beb198839ed.png]

#### 4.3.2 执行SQL ####

首先创建输入表：从Kafka消费数据，其次编写SQL提取字段值，再创建输出表：将数据保存值Hudi表中，最后编写SQL查询Hudi表数据。  
● 第1步、创建输入表，关联Kafka Topic  
– 输入表：Kafka Source

CREATE TABLE order_kafka_source (
      orderId STRING,
      userId STRING,
      orderTime STRING,
      ip STRING,
      orderMoney DOUBLE,
      orderStatus INT
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'order-topic',
      'properties.bootstrap.servers' = 'node1.oldlu.cn:9092',
      'properties.group.id' = 'gid-1001',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json',
      'json.fail-on-missing-field' = 'false',
      'json.ignore-parse-errors' = 'true'
    );
    
    SELECT orderId, userId, orderTime, ip, orderMoney, orderStatus FROM order_kafka_source ;

● 第2步、处理获取Kafka消息数据，提取字段值

SELECT 
      orderId, userId, orderTime, ip, orderMoney, orderStatus, 
      substring(orderId, 0, 17) AS ts, substring(orderTime, 0, 10) AS partition_day 
    FROM order_kafka_source ;

● 第3步、创建输出表，保存数据至Hudi表，设置相关属性  
– 输出表：Hudi Sink

CREATE TABLE order_hudi_sink (
      orderId STRING PRIMARY KEY NOT ENFORCED,
      userId STRING,
      orderTime STRING,
      ip STRING,
      orderMoney DOUBLE,
      orderStatus INT,
      ts STRING,
      partition_day STRING
    )
    PARTITIONED BY (partition_day) 
    WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://node1.oldlu.cn:8020/hudi-warehouse/order_hudi_sink',
      'table.type' = 'MERGE_ON_READ',
      'write.operation' = 'upsert',
      'hoodie.datasource.write.recordkey.field'= 'orderId',
      'write.precombine.field' = 'ts',
      'write.tasks'= '1',
      'compaction.tasks' = '1', 
      'compaction.async.enabled' = 'true', 
      'compaction.trigger.strategy' = 'num_commits', 
      'compaction.delta_commits' = '1'
    );

● 第4步、使用INSERT INTO语句，将数据保存Hudi表  
– 子查询插入INSERT … SELECT …

INSERT INTO order_hudi_sink 
    SELECT
      orderId, userId, orderTime, ip, orderMoney, orderStatus,
      substring(orderId, 0, 17) AS ts, substring(orderTime, 0, 10) AS partition_day 
    FROM order_kafka_source ;
    12345

此时，提交Flink Job运行在FlinkStandalone集群上，示意图如下：  
![在这里插入图片描述][b7c8ff687c5c4c9e9106cb70a7637d9c.png]

只要运行模拟交易订单数据程序，就会将数据发送到Kafka，最后转换保存至Hudi表，截图如下：  
![在这里插入图片描述][482324f23ce24065ad609cbf14a7b9c6.png]

■第5步、编写SELECT语句，查询Hudi表交易订单数据  
– 查询Hudi表数据

> SELECT \* FROM order\_hudi\_sink ;

![在这里插入图片描述][e906aa0b5f1c4c4cbed41d3512338816.png]

## 5 Hudi CDC ##

CDC的全称是Change data Capture，即变更数据捕获，主要面向数据库的变更，是是数据库领域非常常见的技术，主要用于捕获数据库的一些变更，然后可以把变更数据发送到下游。  
![在这里插入图片描述][9a412284285f429fa6f4cc70f9ba8c9d.png]

对于CDC，业界主要有两种类型：一是基于查询的，客户端会通过SQL方式查询源库表变更数据，然后对外发送。二是基于日志，这也是业界广泛使用的一种方式，一般是通过binlog方式，变更的记录会写入binlog，解析binlog后会写入消息系统，或直接基于Flink CDC进行处理。  
![在这里插入图片描述][7c2decf0cf4e4dd09fb2ab2056284098.png]

■基于查询：这种 CDC 技术是入侵式的，需要在数据源执行 SQL 语句。使用这种技术实现CDC 会影响数据源的性能。通常需要扫描包含大量记录的整个表。  
■基于日志：这种 CDC 技术是非侵入性的，不需要在数据源执行 SQL 语句。通过读取源数据库的日志文件以识别对源库表的创建、修改或删除数据。

### 5.1 CDC 数据入湖 ###

基于CDC数据的入湖，这个架构非常简单：上游各种各样的数据源，比如DB的变更数据、事件流，以及各种外部数据源，都可以通过变更流的方式写入表中，再进行外部的查询分析。

![在这里插入图片描述][9da4c241db4f4bf7b028abbd58e780f1.png]

典型CDC入湖的链路：上面的链路是大部分公司采取的链路，前面CDC的数据先通过CDC工具导入Kafka或者Pulsar，再通过Flink或者是Spark流式消费写到Hudi里。第二个架构是通过Flink CDC直联到MySQL上游数据源，直接写到下游Hudi表。  
![在这里插入图片描述][fa57a1baa3bc40bebb70441a4f559b73.png]

### 5.2 Flink CDC Hudi ###

基于Flink CDC技术，实时采集MySQL数据库表数据，进行过ETL转换处理，最终存储Hudi表。  
![在这里插入图片描述][661b8185e78a41f7a0ac8a3cd8415208.png]

#### 5.2.1 业务需求 ####

MySQL数据库创建表，实时添加数据，通过Flink CDC将数据写入Hudi表，并且Hudi与Hive集成，自动在Hive中创建表与添加分区信息，最后Hive终端Beeline查询分析数据。  
![在这里插入图片描述][c14323e57db04c43bdb0b7fd63aacd7d.png]

Hudi 表与Hive表，自动关联集成，需要重新编译Hudi源码，指定Hive版本及编译时包含Hive依赖jar包，具体步骤如下。  
● 修改Hudi集成flink和Hive编译依赖版本配置  
原因：现在版本Hudi，在编译的时候本身默认已经集成的flink-SQL-connector-hive的包，会和Flink lib包下的flink-SQL-connector-hive冲突。所以，编译的过程中只修改hive编译版本。  
文件：hudi-0.9.0/packaging/hudi-flink-bundle/pom.xml  
![在这里插入图片描述][f6e2f0cb791c43c29654981483dd2183.png]

● 编译Hudi源码

> mvn clean install -DskipTests -Drat.skip=true -Dscala-2.12 -Dspark3  
> \-Pflink-bundle-shade-hive2

编译完成以后，有2个jar包，至关重要：  
 hudi-flink-bundle\_2.12-0.9.0.jar，位于  
hudi-0.9.0/packaging/hudi-flink-bundle/target，flink 用来写入和读取数据，将其拷贝至  
KaTeX parse error: Unexpected character: '' at position 39: …同名jar包，先删除再拷贝。 ̲ hudi-hadoop-mr…HIVE\_HOME/lib目录中。  
■ 将Flink CDC MySQL对应jar包，放到$FLINK\_HOME/lib目录中  
flink-sql-connector-mysql-cdc-1.3.0.jar

至此，$FLINK\_HOME/lib目录中，有如下所需的jar包，缺一不可，注意版本号。  
![在这里插入图片描述][20f9f5f983c448ee9067d10527c3402a.png]

#### 5.2.2 创建 MySQL 表 ####

首先开启MySQL数据库binlog日志，再重启MySQL数据库服务，最后创建表。  
■第一步、开启MySQL binlog日志

> \[root@node1 ~\]\# vim /etc/my.cnf

在\[mysqld\]下面添加内容：

server-id=2
    log-bin=mysql-bin
    binlog_format=row
    expire_logs_days=15
    binlog_row_image=full

![在这里插入图片描述][9342d0aea0d6407496de928b09cdee71.png]

■第二步、重启MySQL Server

> service mysqld restart

登录MySQL Client命令行，查看是否生效。  
![在这里插入图片描述][5721ab628a0449a0a8456b076bbb016c.png]

■第三步、在MySQL数据库，创建表  
– MySQL 数据库创建表

create database test ;
    create table test.tbl_users(
       id bigint auto_increment primary key,
       name varchar(20) null,
       birthday timestamp default CURRENT_TIMESTAMP not null,
       ts timestamp default CURRENT_TIMESTAMP not null
    );

![在这里插入图片描述][cee43f841ed840a7b656398c79adbcbc.png]

#### 5.2.3 创建 CDC 表 ####

先启动HDFS服务、Hive MetaStore和HiveServer2服务和Flink Standalone集群，再运行SQL Client，最后创建表关联MySQL表，采用MySQL CDC方式。  
● 启动HDFS服务，分别启动NameNode和DataNode  
– 启动HDFS服务

> hadoop-daemon.sh start namenode  
> hadoop-daemon.sh start datanode

● 启动Hive服务：元数据MetaStore和HiveServer2  
– Hive服务

> /export/server/hive/bin/start-metastore.sh  
> /export/server/hive/bin/start-hiveserver2.sh

■启动Flink Standalone集群  
– 启动Flink Standalone集群

> export HADOOP\_CLASSPATH=`/export/server/hadoop/bin/hadoop classpath`  
> /export/server/flink/bin/start-cluster.sh

■启动SQL Client客户端

> /export/server/flink/bin/sql-client.sh embedded -j  
> /export/server/flink/lib/hudi-flink-bundle\_2.12-0.9.0.jar shell

设置属性：

> set execution.result-mode=tableau; set  
> execution.checkpointing.interval=3sec;

● 创建输入表，关联MySQL表，采用MySQL CDC 关联  
– Flink SQL Client创建表

CREATE TABLE users_source_mysql (
      id BIGINT PRIMARY KEY NOT ENFORCED,
      name STRING,
      birthday TIMESTAMP(3),
      ts TIMESTAMP(3)
    ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'node1.oldlu.cn',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'server-time-zone' = 'Asia/Shanghai',
    'debezium.snapshot.mode' = 'initial',
    'database-name' = 'test',
    'table-name' = 'tbl_users'
    );

查询表的结构，其中id为主键，ts为数据合并字段。  
![在这里插入图片描述][0b973fabae6947f999c064e61995c930.png]

● 查询CDC表数据  
– 查询数据

> select \* from users\_source\_mysql;

![在这里插入图片描述][84e93cc31b854c29921d6b4c40711de2.png]

● 开启MySQL Client客户端，执行DML语句，插入数据

> insert into test.tbl\_users (name) values (‘zhangsan’) insert into  
> test.tbl\_users (name) values (‘lisi’); insert into test.tbl\_users  
> (name) values (‘wangwu’); insert into test.tbl\_users (name) values  
> (‘laoda’); insert into test.tbl\_users (name) values (‘laoer’);

#### 5.2.4 创建视图 ####

创建一个临时视图，增加分区列part，方便后续同步hive分区表。  
– 创建一个临时视图，增加分区列 方便后续同步hive分区表

> create view view\_users\_cdc AS SELECT \*, DATE\_FORMAT(birthday,  
> ‘yyyyMMdd’) as part FROM users\_source\_mysql;

查看视图view中数据

> select \* from view\_users\_cdc;

![在这里插入图片描述][c90e63a7eae8448fb33c8e6a0924d93a.png]

#### 5.2.5 创建 Hudi 表 ####

创建 CDC Hudi Sink表，并自动同步hive分区表，具体DDL语句。

CREATE TABLE users_sink_hudi_hive(
    id bigint ,
    name string,
    birthday TIMESTAMP(3),
    ts TIMESTAMP(3),
    part VARCHAR(20),
    primary key(id) not enforced
    )
    PARTITIONED BY (part)
    with(
    'connector'='hudi',
    'path'= 'hdfs://node1.oldlu.cn:8020/ehualu/hudi-warehouse/users_sink_hudi_hive', 
    'table.type'= 'MERGE_ON_READ',
    'hoodie.datasource.write.recordkey.field'= 'id', 
    'write.precombine.field'= 'ts',
    'write.tasks'= '1',
    'write.rate.limit'= '2000', 
    'compaction.tasks'= '1', 
    'compaction.async.enabled'= 'true',
    'compaction.trigger.strategy'= 'num_commits',
    'compaction.delta_commits'= '1',
    'changelog.enabled'= 'true',
    'read.streaming.enabled'= 'true',
    'read.streaming.check-interval'= '3',
    'hive_sync.enable'= 'true',
    'hive_sync.mode'= 'hms',
    'hive_sync.metastore.uris'= 'thrift://node1.oldlu.cn:9083',
    'hive_sync.jdbc_url'= 'jdbc:hive2://node1.oldlu.cn:10000',
    'hive_sync.table'= 'users_sink_hudi_hive',
    'hive_sync.db'= 'default',
    'hive_sync.username'= 'root',
    'hive_sync.password'= '123456',
    'hive_sync.support_timestamp'= 'true'
    );

此处Hudi表类型：MOR，Merge on Read （读时合并），快照查询+增量查询+读取优化查询（近实时）。使用列式存储（parquet）+行式文件（arvo）组合存储数据。更新记录到增量文件中，然后进行同步或异步压缩来生成新版本的列式文件。  
![在这里插入图片描述][03997e0b876f44de83870888e2a1c1f1.png]

#### 5.2.6 数据写入Hudi表 ####

编写INSERT语句，从视图中查询数据，再写入Hudi表中，语句如下：

> insert into users\_sink\_hudi\_hive select id, name, birthday, ts, part  
> from view\_users\_cdc;

Flink web UI DAG图：  
![在这里插入图片描述][725a40fab3d04922b1f443cf3528e50d.png]

HDFS上Hudi文件目录情况：  
![在这里插入图片描述][1ca3942de5da49d3b0677483f5af9501.png]

查询Hudi表数据，SELECT语句如下：

> select \* from users\_sink\_hudi\_hive;

![在这里插入图片描述][3b3f99bcac0b4f0d900c1f2d28317493.png]

#### 5.2.7 Hive 表查询 ####

需要引入hudi-hadoop-mr-bundle-0.9.0.jar包，放到$HIVE\_HOME/lib下。  
![在这里插入图片描述][361c2673037f4873881b992014afda01.png]

启动Hive中beeline客户端，连接HiveServer2服务：

> /export/server/hive/bin/beeline -u jdbc:hive2://node1.oldlu.cn:10000  
> \-n root -p 123456

![在这里插入图片描述][600b5947f7a54eb69b43f51d35f1fffa.png]

已自动生产hudi MOR模式的2张表：  
■users\_sink\_hudi\_hive\_ro，ro 表全称 read oprimized table，对于 MOR 表同步的 xxx\_ro 表，只暴露压缩后的 parquet。其查询方式和COW表类似。设置完 hiveInputFormat 之后 和普通的 Hive 表一样查询即可；  
■users\_sink\_hudi\_hive\_rt，rt表示增量视图，主要针对增量查询的rt表；ro表只能查parquet文件数据， rt表 parquet文件数据和log文件数据都可查；  
查看自动生成表users\_sink\_hudi\_hive\_ro结构：

CREATE EXTERNAL TABLE `users_sink_hudi_hive_ro`(
      `_hoodie_commit_time` string COMMENT '', 
      `_hoodie_commit_seqno` string COMMENT '', 
      `_hoodie_record_key` string COMMENT '', 
      `_hoodie_partition_path` string COMMENT '', 
      `_hoodie_file_name` string COMMENT '', 
      `_hoodie_operation` string COMMENT '', 
      `id` bigint COMMENT '', 
      `name` string COMMENT '', 
      `birthday` bigint COMMENT '', 
      `ts` bigint COMMENT '')
    PARTITIONED BY ( 
      `part` string COMMENT '')
    ROW FORMAT SERDE 
      'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
    WITH SERDEPROPERTIES ( 
      'hoodie.query.as.ro.table'='true', 
      'path'='hdfs://node1.oldlu.cn:8020/users_sink_hudi_hive') 
    STORED AS INPUTFORMAT 
      'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
    OUTPUTFORMAT 
      'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
    LOCATION
      'hdfs://node1.oldlu.cn:8020/users_sink_hudi_hive'
    TBLPROPERTIES (
      'last_commit_time_sync'='20211125095818', 
      'spark.sql.sources.provider'='hudi', 
      'spark.sql.sources.schema.numPartCols'='1', 
      'spark.sql.sources.schema.numParts'='1', 
    'spark.sql.sources.schema.part.0'='{\"type\":\"struct\",\"fields\":[{\"name\":\"_hoodie_commit_time\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"_hoodie_commit_seqno\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"_hoodie_record_key\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"_hoodie_partition_path\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"_hoodie_file_name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"_hoodie_operation\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"id\",\"type\":\"long\",\"nullable\":false,\"metadata\":{}},{\"name\":\"name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"birthday\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ts\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}},{\"name\":\"part\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}', 
      'spark.sql.sources.schema.partCol.0'='partition', 
      'transient_lastDdlTime'='1637743860')

查看自动生成表的分区信息：

> show partitions users\_sink\_hudi\_hive\_ro ; show partitions  
> users\_sink\_hudi\_hive\_rt ;

![在这里插入图片描述][f7b214bf03194138bb0f7287a72ba039.png]

查询Hive 分区表数据

> set hive.exec.mode.local.auto=true; set hive.input.format =  
> org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat; set  
> hive.mapred.mode=nonstrict ;
> 
> select id, name, birthday, ts, `part` from users\_sink\_hudi\_hive\_ro;

![在这里插入图片描述][c086d475af514484aa4dc6297b58975d.png]

指定分区字段过滤，查询数据

> select name, ts from users\_sink\_hudi\_hive\_ro where part =‘20211125’;  
> select name, ts from users\_sink\_hudi\_hive\_rt where part =‘20211125’;

![在这里插入图片描述][7a3b9d33cd114e08b6ffa169a8e9d016.png]

### 5.3 Hudi Client操作Hudi表 ###

进入Hudi客户端命令行：hudi-0.9.0/hudi-cli/hudi-cli.sh  
![在这里插入图片描述][c60f8079ddce4f1db0a69e991d494909.png]

连接Hudi表，查看表信息

> connect --path hdfs://node1.oldlu.cn:8020/users\_sink\_hudi\_hive

![在这里插入图片描述][d60254571f6741e2b1f6ec3fafd3da90.png]

查看Hudi commit信息

> commits show --sortBy “CommitTime”

![在这里插入图片描述][bc0682abed53426eb8945db8bc4b331c.png]  
查看Hudi compactions 计划

> compactions show all  
> ![在这里插入图片描述][7d23fce9a19e4117ba0fbda6531b15c7.png]

[1eee962b1eaa4b19a81ff16feb3aff68.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/4d34dc9279cd4c1889c420f0b0b4bbec.png
[545f480c34384e04bc2c5c59b08184c2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/a1db0f6cb03f44719849679909ea6c92.png
[325e194e817a40f08eb69deda1ffd444.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b01541c1c8644a1a83bd12214e00edb2.png
[884cd38d98d3484790cd8aebeba881d9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/9643a86babc54b1992eb396d3bdbb8b9.png
[68accaf67cab401789d95e5940ee6a78.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/de00384390b64376b7d96f2edf1a9303.png
[41b0ada4afd042a2bb72e80cd10c9536.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/35b595e03803484da65f7a0695bb853b.png
[07a279397d704dc885f506cf43bdf7f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/8ace8d12a98a4ef0b586951755a401d8.png
[c3589734b25a4b4b892f6892ab157468.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/bb74c924c159473488b90c0bf942bf0f.png
[ff4af9957c3b42cebbf4f31dc9433111.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/5cae112765a0452cab8c6f33d75557a3.png
[6a259a51a212418abd540dc08c28faca.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/a59ca07268fe4044ae6819ddf011e636.png
[89ebd7441d624718904c24c5a188c951.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/228b052e12874ffb808b28164e5467ff.png
[dc3e3fccc3094a4eb2b2e03fa57ea4e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/2b14f396b02d45c8bff814242a97235f.png
[b84d6134fc51415881ad0c136632f944.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/600f2a2cab18455da6a718aa96687736.png
[406f56c1a880446390be3d60b87e9c1a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/5833debed12849cd85f4d466b757125a.png
[f37746ea808946e79126869a3ff2143f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b4d23a6fe164422f8355852a398f7cb5.png
[0af16cf4db1d4508b8651f0ce46839d6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/8d406935847548a683bac088a1fab319.png
[73b2b4b1beb049da868f50ac1e9e7b79.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/12f688288c4048f79b5b69211b380e8b.png
[77fcfc6ef4bf489fa2fb96ca44fcf38f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/1ec6adc5f986428a9fd44d8fede71fab.png
[60cb1771ead347708d865f095205d27a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/4b8f14b80d0e40429b557e7f5bd97d8e.png
[9176e21665bc4f9b9d518c4f4d44240a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/002faaf5c139447eb0bc6ff04393103e.png
[914a3f2b49174aa9abbb790fe8c5e076.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/aeb64bfe8b9a41868f5c093ec93cc11b.png
[1d2c41ef02f7464db0261fac0a3344c2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/663e07be48c843b4804243165cd63398.png
[75ab8e15c9b24eda91107e9f2c27d690.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/95131f8de08e420895f8d6c27a294bc8.png
[41ade2f3297c48a1bf902fbe3cbdc526.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/bcec1ae8fbb942b0b30bb0ca4502c5b6.png
[de7a3e2eb4ca423d9c143b74dacf346d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/5f6b498403b94a53941d5ca33d717477.png
[913a15cf9d044ea1a80be611a751ea8c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/03a5bc3cd441491199635e6dfcf72ac4.png
[5c66f362c5e34fa29f62a49e787593c7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/74cf3b3d1d4f4fdc87aaf2d9ff6dce7b.png
[9ae9108a3cb24a69b5e2e941cd90efe4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/79b0a8b373c14cb18ca903c3e345285c.png
[6c9058d04f734d25a3c55baa73374455.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/429610e59f1c4ecd98121ac66e258f6a.png
[a60f851aecd2402c89dcb27a937ed084.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/5de0bc904ddd4499a88016bc6670ad78.png
[48d12fd944a047918162ac794fca909c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/93ff09168b5442e1b238a5ece2b8663a.png
[bef411c891b840d6b621fa9ac211e282.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/89b6386303164d839e3edc3bd4dcdb2f.png
[f9b58abdd31f4bcd9106d0c0c0ad0986.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/dac9e44c654647919bfd9dcc558c653c.png
[caad7bef601348f7961badb51955c9d3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/1d9018e435bb4965934b48f6dea50bcb.png
[75615a0034e447ddaae64beb198839ed.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/c42b026f972b404ba6aa7205429567bf.png
[b7c8ff687c5c4c9e9106cb70a7637d9c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/a6eb2694679b4d9381738a581ee61243.png
[482324f23ce24065ad609cbf14a7b9c6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/905f0e39bf1145a29d487b1acb5b4a52.png
[e906aa0b5f1c4c4cbed41d3512338816.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d059ad5d80174cbdb111de40aca12f79.png
[9a412284285f429fa6f4cc70f9ba8c9d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/219db33d4b334a31b9f7bddca84b4206.png
[7c2decf0cf4e4dd09fb2ab2056284098.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/e9f94a28416f4a7eb62a2e29027afa0e.png
[9da4c241db4f4bf7b028abbd58e780f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/14369911d94c4b5d9b42ae6843ee67cd.png
[fa57a1baa3bc40bebb70441a4f559b73.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/201387ae8d8641df919be56bfad4ed74.png
[661b8185e78a41f7a0ac8a3cd8415208.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/ab54633b27224a54be99502f5261188b.png
[c14323e57db04c43bdb0b7fd63aacd7d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/9e646a8ac5124c08b571c627cd8621f4.png
[f6e2f0cb791c43c29654981483dd2183.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/ea47ad32de1d42c4947342d551ed728e.png
[20f9f5f983c448ee9067d10527c3402a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d5910c222c3b481fa16251c12dbc24b8.png
[9342d0aea0d6407496de928b09cdee71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/2e7bafecaac7491c97dd3fc563ea890d.png
[5721ab628a0449a0a8456b076bbb016c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/ea7adb490afb471ab9ce6f966d5b4fe7.png
[cee43f841ed840a7b656398c79adbcbc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/c74b9cf90e4d411da920a1c9dcda06c7.png
[0b973fabae6947f999c064e61995c930.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d70549fe7fcc40c0860809167cbca41f.png
[84e93cc31b854c29921d6b4c40711de2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b5ba669be35a4e7293c8cddd7b070d7a.png
[c90e63a7eae8448fb33c8e6a0924d93a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d5a6f2ff4967447fb8cd628a59cec60f.png
[03997e0b876f44de83870888e2a1c1f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/585e9777abed411ab3372d0a6b71348a.png
[725a40fab3d04922b1f443cf3528e50d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/b45417181c3642d783a1d1e697b3a888.png
[1ca3942de5da49d3b0677483f5af9501.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/a12b6395ff6942da9076d21727e44018.png
[3b3f99bcac0b4f0d900c1f2d28317493.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/4f7892c74ec54232bd974f7c9f507d1e.png
[361c2673037f4873881b992014afda01.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/d33a2bc5a44b4d2c9b73e99b520b7de4.png
[600b5947f7a54eb69b43f51d35f1fffa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/e02a3d75704b48c4ad9f24b0ea049974.png
[f7b214bf03194138bb0f7287a72ba039.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3c425b098bb949669e1ff0e37a00af8f.png
[c086d475af514484aa4dc6297b58975d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/91b97f69ae13423d86cd2242313a0e10.png
[7a3b9d33cd114e08b6ffa169a8e9d016.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/e6197d230f9f436796efc20dfd6bd0e5.png
[c60f8079ddce4f1db0a69e991d494909.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/355d3f7892704f82b0af5f32c2dacbe2.png
[d60254571f6741e2b1f6ec3fafd3da90.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/94b33f62880049b0b244c93ddc6d2c8f.png
[bc0682abed53426eb8945db8bc4b331c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/30821c15b81a492fb29f8dcee10323d1.png
[7d23fce9a19e4117ba0fbda6531b15c7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/ccb6fc10161b4dab96a2f8d693ae59d9.png