ToB企服应用市场:ToB评测及商务社交产业平台

标题: 30分钟掌握沧湖一体化：flink+hudi（干货，建议收藏）_flink hudi sink [打印本页]

---

作者: 农妇山泉一亩田    时间: 2024-10-22 19:16
标题: 30分钟掌握沧湖一体化：flink+hudi（干货，建议收藏）_flink hudi sink

• 索引，将传入的记录键快速映射到文件（如果已存在记录键）。索引实现是可插拔的，Bloom过滤器-由于不依赖任何外部系统，因此它是默认配置，索引和数据始终保持同等。Apache HBase-对少量key更高效。在索引标记过程中大概会节省几秒钟。
  
• 数据，Hudi以两种差别的存储格式存储数据。实际使用的格式是可插入的，但要求具有以下特性–读优化的列存储格式（ROFormat），默认值为Apache Parquet；写优化的基于行的存储格式（WOFormat），默认值为Apache Avro。
  

为什么Hudi对于大规模和近实时应用很紧张？

Hudi办理了以下限制

• HDFS的可伸缩性限制
  
• 需要在Hadoop中更快地呈现数据
  
• 没有直接支持对现有数据的更新和删除
  
• 快速的ETL和建模
  
• 要检索全部更新的记录，无论这些更新是添加到最近日期分区的新记录还是对旧数据的更新，Hudi都允许用户使用最后一个检查点时间戳。此过程不用执行扫描整个源表的查询
  

Hudi的优势

• HDFS中的可伸缩性限制。
  
• Hadoop中数据的快速呈现
  
• 支持对于现有数据的更新和删除
  
• 快速的ETL和建模
  

（以上内容主要引用于：Apache Hudi 详解_风中云彩的博客-CSDN博客）
新架构与湖仓一体

通过湖仓一体、流批一体，准实时场景下做到了：数据同源、同计算引擎、同存储、同计算口径。数据的时效性可以到分钟级，能很好的满足业务准实时数仓的需求。下面是架构图：

MySQL 数据通过 Flink CDC 进入到 Kafka。之以是数据先入 Kafka 而不是直接入 Hudi，是为了实现多个实时使命复用 MySQL 过来的数据，避免多个使命通过 Flink CDC 接 MySQL 表以及 Binlog，对 MySQL 库的性能造成影响。
通过 CDC 进入到 Kafka 的数据除了落一份到离线数据堆栈的 ODS 层之外，会同时按照实时数据堆栈的链路，从 ODS->DWD->DWS->OLAP 数据库，最后供报表等数据服务使用。实时数仓的每一层结果数据会准实时的落一份到离线数仓，通过这种方式做到步伐一次开辟、指标口径统一，数据统一。
从架构图上，可以看到有一步数据修正 (重跑汗青数据) 的动作，之以是有这一步是思量到：有大概存在由于口径调解大概前一天的实时使命计算结果错误，导致重跑汗青数据的情况。
而存储在 Kafka 的数据有失效时间，不会存太久的汗青数据，重跑好久的汗青数据无法从 Kafka 中获取汗青源数据。再者，如果把大量的汗青数据再一次推到 Kafka，走实时计算的链路来修正汗青数据，大概会影响当天的实时作业。以是针对重跑汗青数据，会通过数据修正这一步来处置惩罚。
总体上说，这个架构属于 Lambda 和 Kappa 混搭的架构。流批一体数据堆栈的各个数据链路有数据质量校验的流程。第二天对前一天的数据进行对账，如果前一天实时计算的数据无非常，则不需要修正数据，Kappa 架构已经充足。
（本节内容，引用自：37 手游基于 Flink CDC + Hudi 湖仓一体方案实践）
最佳实践

版本搭配

版本选择，这个问题大概会成为困扰大家的第一个绊脚石，下面是hudi中文社区推荐的版本适配：
flinkhudi1.12.20.9.01.13.10.10.0 官方说的支持版本是这样， 不过如今我的1.13和0.10组合并没有配置成功，以是大家还是只管选择 1.12.2+0.9.0 吧，配合scala 2.11 。
下载hudi

   https://mvnrepository.com/artifact/org.apache.hudi/hudi-flink-bundle
  执行

如果将 hudi-flink-bundle_2.11-0.9.0.jar 放到了 flink/lib 下，则只需要如下执行即可，否则会出现各种找不到类的非常
   bin/sql-client.sh embedded
  Flink on hudi

新建maven工程，修改pom如下

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.    <modelVersion>4.0.0</modelVersion>
6. ​
7.    <groupId>org.example</groupId>
8.    <artifactId>flink_hudi_test</artifactId>
9.    <version>1.0-SNAPSHOT</version>
10. ​
11.    <properties>
12.        <maven.compiler.source>8</maven.compiler.source>
13.        <maven.compiler.target>8</maven.compiler.target>
14.    </properties>
15. ​
16.    <dependencies>
17. ​
18. ​
19.        <dependency>
20.            <groupId>org.apache.hadoop</groupId>
21.            <artifactId>hadoop-client</artifactId>
22.            <version>2.9.2</version>
23.        </dependency>
24.        <dependency>
25.            <groupId>org.apache.hadoop</groupId>
26.            <artifactId>hadoop-hdfs</artifactId>
27.            <version>2.9.2</version>
28.        </dependency>
29.        <dependency>
30.            <groupId>org.apache.hadoop</groupId>
31.            <artifactId>hadoop-common</artifactId>
32.            <version>2.9.2</version>
33.        </dependency>
34. ​
35. ​
36.        <dependency>
37.            <groupId>org.apache.flink</groupId>
38.            <artifactId>flink-core</artifactId>
39.            <version>1.12.2</version>
40.        </dependency>
41.        <dependency>
42.            <groupId>org.apache.flink</groupId>
43.            <artifactId>flink-streaming-java_2.11</artifactId>
44.            <version>1.12.2</version>
45.        </dependency>
46. ​
47.        <!--        <dependency>-->
48.        <!--            <groupId>org.apache.flink</groupId>-->
49.        <!--            <artifactId>flink-jdbc_2.12</artifactId>-->
50.        <!--            <version>1.10.3</version>-->
51.        <!--        </dependency>-->
52.        <dependency>
53.            <groupId>org.apache.flink</groupId>
54.            <artifactId>flink-connector-jdbc_2.11</artifactId>
55.            <version>1.12.2</version>
56.        </dependency>
57. ​
58.        <dependency>
59.            <groupId>org.apache.flink</groupId>
60.            <artifactId>flink-java</artifactId>
61.            <version>1.12.2</version>
62.        </dependency>
63.        <dependency>
64.            <groupId>org.apache.flink</groupId>
65.            <artifactId>flink-clients_2.11</artifactId>
66.            <version>1.12.2</version>
67.        </dependency>
68.        <dependency>
69.            <groupId>org.apache.flink</groupId>
70.            <artifactId>flink-table-api-java-bridge_2.11</artifactId>
71.            <version>1.12.2</version>
72.        </dependency>
73.        <dependency>
74.            <groupId>org.apache.flink</groupId>
75.            <artifactId>flink-table-common</artifactId>
76.            <version>1.12.2</version>
77.        </dependency>
78. ​
79.        <dependency>
80.            <groupId>org.apache.flink</groupId>
81.            <artifactId>flink-table-planner_2.11</artifactId>
82.            <version>1.12.2</version>
83.        </dependency>
84. ​
85.        <dependency>
86.            <groupId>org.apache.flink</groupId>
87.            <artifactId>flink-table-planner-blink_2.11</artifactId>
88.            <version>1.12.2</version>
89.        </dependency>
90.        <dependency>
91.            <groupId>org.apache.flink</groupId>
92.            <artifactId>flink-table-planner-blink_2.11</artifactId>
93.            <version>1.12.2</version>
94.            <type>test-jar</type>
95.        </dependency>
96. ​
97.        <dependency>
98.            <groupId>com.alibaba.ververica</groupId>
99.            <artifactId>flink-connector-mysql-cdc</artifactId>
100.            <version>1.2.0</version>
101.        </dependency>
102. ​
103. ​
104. <!--        <dependency>-->
105. <!--            <groupId>org.apache.hudi</groupId>-->
106. <!--            <artifactId>hudi-flink-client</artifactId>-->
107. <!--            <version>0.9.0</version>-->
108. <!--        </dependency>-->
109. <!--        <dependency>-->
110. <!--            <groupId>org.apache.hudi</groupId>-->
111. <!--            <artifactId>hudi-common</artifactId>-->
112. <!--            <version>0.9.0</version>-->
113. <!--        </dependency>-->
114. ​
115. <!--        <dependency>-->
116. <!--            <groupId>org.apache.hudi</groupId>-->
117. <!--            <artifactId>hudi-hadoop-mr-bundle</artifactId>-->
118. <!--            <version>0.9.0</version>-->
119. <!--        </dependency>-->
120. ​
121. ​
122.        <dependency>
123.            <groupId>org.apache.hudi</groupId>
124.            <artifactId>hudi-flink-bundle_2.11</artifactId>
125.            <version>0.9.0</version>
126.        </dependency>
127. ​
128.        <dependency>
129.            <groupId>mysql</groupId>
130.            <artifactId>mysql-connector-java</artifactId>
131.            <version>5.1.49</version>
132.        </dependency>
133. ​
134. ​
135.    </dependencies>
136. </project>

复制代码

我们通过构建查询insert into t2 select replace(uuid(),'-',''),id,name,description,now() from mysql_binlog 将创建的mysql表，插入到hudi里。

1. package name.lijiaqi;
2. ​
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. import org.apache.flink.table.api.EnvironmentSettings;
5. import org.apache.flink.table.api.SqlDialect;
6. import org.apache.flink.table.api.TableResult;
7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
8. ​
9. public class MysqlToHudiExample {
10.    public static void main(String[] args) throws Exception {
11.        EnvironmentSettings fsSettings = EnvironmentSettings.newInstance()
12.                .useBlinkPlanner()
13.                .inStreamingMode()
14.                .build();
15.        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.        env.setParallelism(1);
17.        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings);
18. ​
19.        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
20. ​
21.        // 数据源表
22.        String sourceDDL =
23.                "CREATE TABLE mysql_binlog (\n" +
24.                        " id INT NOT NULL,\n" +
25.                        " name STRING,\n" +
26.                        " description STRING\n" +
27.                        ") WITH (\n" +
28.                        " 'connector' = 'jdbc',\n" +
29.                        " 'url' = 'jdbc:mysql://127.0.0.1:3306/test', \n"+
30.                        " 'driver' = 'com.mysql.jdbc.Driver', \n"+
31.                        " 'username' = 'root',\n" +
32.                        " 'password' = 'dafei1288', \n" +
33.                        " 'table-name' = 'test_cdc'\n" +
34.                        ")";
35. ​
36.        // 输出目标表
37.        String sinkDDL =
38.                "CREATE TABLE t2(\n" +
39.                        "\tuuid VARCHAR(20),\n"+
40.                        "\tid INT NOT NULL,\n" +
41.                        "\tname VARCHAR(40),\n" +
42.                        "\tdescription VARCHAR(40),\n" +
43.                        "\tts TIMESTAMP(3)\n"+
44. //                        "\t`partition` VARCHAR(20)\n" +
45.                        ")\n" +
46. //                        "PARTITIONED BY (`partition`)\n" +
47.                        "WITH (\n" +
48.                        "\t'connector' = 'hudi',\n" +
49.                        "\t'path' = 'hdfs://172.19.28.4:9000/hudi_t4/',\n" +
50.                        "\t'table.type' = 'MERGE_ON_READ'\n" +
51.                        ")" ;
52.        // 简单的聚合处理
53.        String transformSQL =
54.                "insert into t2 select replace(uuid(),'-',''),id,name,description,now()  from mysql_binlog";
55. ​
56.        tableEnv.executeSql(sourceDDL);
57.        tableEnv.executeSql(sinkDDL);
58.        TableResult result = tableEnv.executeSql(transformSQL);
59.        result.print();
60. ​
61.        env.execute("mysql-to-hudi");
62.    }
63. }

复制代码

查询hudi

1. package name.lijiaqi;
2. ​
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. import org.apache.flink.table.api.EnvironmentSettings;
5. import org.apache.flink.table.api.SqlDialect;
6. import org.apache.flink.table.api.TableResult;
7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
8. ​
9. public class ReadHudi {
10.    public static void main(String[] args) throws Exception {
11.        EnvironmentSettings fsSettings = EnvironmentSettings.newInstance()
12.                .useBlinkPlanner()
13.                .inStreamingMode()
14.                .build();
15.        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.        env.setParallelism(1);
17.        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings);
18. ​
19.        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
20. ​
21.        String sourceDDL =
22.                "CREATE TABLE t2(\n" +
23.                        "\tuuid VARCHAR(20),\n"+
24.                        "\tid INT NOT NULL,\n" +
25.                        "\tname VARCHAR(40),\n" +
26.                        "\tdescription VARCHAR(40),\n" +
27.                        "\tts TIMESTAMP(3)\n"+
28. //                        "\t`partition` VARCHAR(20)\n" +
29.                        ")\n" +
30. //                        "PARTITIONED BY (`partition`)\n" +
31.                        "WITH (\n" +
32.                        "\t'connector' = 'hudi',\n" +
33.                        "\t'path' = 'hdfs://172.19.28.4:9000/hudi_t4/',\n" +
34.                        "\t'table.type' = 'MERGE_ON_READ'\n" +
35.                        ")" ;
36.        tableEnv.executeSql(sourceDDL);
37.        TableResult result2 = tableEnv.executeSql("select * from t2");
38.        result2.print();
39. ​
40.        env.execute("read_hudi");
41.    }
42. }

复制代码

展示结果

Flink CDC 2.0 on Hudi

添加依赖

添加如下依赖到$FLINK_HOME/lib下

• hudi-flink-bundle_2.11-0.10.0-SNAPSHOT.jar (修改 Master 分支的 Hudi Flink 版本为 1.13.2 然后构建)
  
• hadoop-mapreduce-client-core-2.7.3.jar (办理 Hudi ClassNotFoundException)
  
• flink-sql-connector-mysql-cdc-2.0.0.jar
  
• flink-format-changelog-json-2.0.0.jar
  
• flink-sql-connector-kafka_2.11-1.13.2.jar
  

留意，在探求jar的时候，cdc 2.0 更新过group id ，不再试 com.alibaba.ververica 而是改成了 com.ververica
flink sql cdc on hudi

创建mysql cdc表

1. CREATE  TABLE mysql_users (
2.     id BIGINT PRIMARY KEY NOT ENFORCED ,
3.     name STRING,
4.     birthday TIMESTAMP(3),
5.     ts TIMESTAMP(3)
6. ) WITH (
7.     'connector' = 'mysql-cdc',
8.     'hostname' = 'localhost',
9.     'port' = '3306',
10.     'username' = 'root',
11.     'password' = 'dafei1288',
12.     'server-time-zone' = 'Asia/Shanghai',
13.     'database-name' = 'test',
14.     'table-name' = 'users'  
15. );
18. # **写在最后**
20. 为了这次面试，也收集了很多的面试题！
22. 以下是部分面试题截图
24. 
25. Y NOT ENFORCED ,
26.     name STRING,
27.     birthday TIMESTAMP(3),
28.     ts TIMESTAMP(3)
29. ) WITH (
30.     'connector' = 'mysql-cdc',
31.     'hostname' = 'localhost',
32.     'port' = '3306',
33.     'username' = 'root',
34.     'password' = 'dafei1288',
35.     'server-time-zone' = 'Asia/Shanghai',
36.     'database-name' = 'test',
37.     'table-name' = 'users'  
38. );
41. # **写在最后**
43. 为了这次面试，也收集了很多的面试题！
45. 以下是部分面试题截图
47. [外链图片转存中...(img-eEI9zNAD-1723525790091)]

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4