基于 flink rest api 的 flink web admin,
支持任务持久化管理，JOB编辑、提交、启动、停止。
支持多集群发布任务，任务根据场景选择不同的集群。
支持手动Savepoint及指定Savepoint恢复任务。
支持数据源管理，加密保存数据源配置信息。
支持可视化的CDC数据同步任务，可视化配置单表，多表，全量及增量同步。 、

Flink1.13.2三种方式安装部署

https://www.aboutyun.com/forum.php?mod=viewthread&tid=31630&fromguid=hot

flink-admin 控制台

https://gitee.com/chongho/flink-admin.git

整合Apache Hudi+Mysql+FlinkCDC2.1+CDH6.3.0

一、环境准备

1.环境准备：

flink 1.13.1+hudi0.10+hive2.1.1+cdh6.3.0+mysql5.7+flinkcdc2.1+flink web平台

二.编译hudi（这个编译是以前的一个测试版本，编译大同小异）

1.使用git命令下载hudi0.10的代码

  1. steven@wangyuxiangdeMacBook-Pro  ~  git clone  https://github.com/apache/hudi.git
  2. Cloning into 'hudi'...
  3. remote: Enumerating objects: 122696, done.
  4. remote: Counting objects: 100% (5537/5537), done.
  5. remote: Compressing objects: 100% (674/674), done.
  6. remote: Total 122696 (delta 4071), reused 4988 (delta 3811), pack-reused 117159
  7. Receiving objects: 100% (122696/122696), 75.85 MiB | 5.32 MiB/s, done.
  8. Resolving deltas: 100% (61608/61608), done.

image.png
2.使用idea打开hudi更改packging—hudi-flink-bundle下的pom.xml,更改flink-bundel-shade-hive2下的hive-version更改为chd6.3.0的版本。
image.png
3.使用命令进行编译

  1. mvn clean install -DskipTests -DskipITs -Dcheckstyle.skip=true -Drat.skip=true -Dhadoop.version=3.0.0  -Pflink-bundle-shade-hive2
  2. 1.因为chd6.3.0使用的是hadoop3.0.0,所以要指定hadoop的版本
  3. 2.使用的是hive2.1.1的版本，也要指定hive的版本，不然使用sync to hive的时候，会报类的冲突问题。

image.png
image.png
出现以上的证明编译成功。
5.在packaging下面各个组件中有编译好的jar包。
image.png
image.png
image.png

6.部署同步sync to hive的环境
将hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar包放入到以下路径
路径如下：

  1. [flink@dbos-bigdata-test005 jars]$ pwd
  2. /opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/jars

image.png
image.png

  1. 进入到hive的lib路径,每一台hive节点都要放
  2. [flink@dbos-bigdata-test005 lib]$ pwd
  3. /opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/hive/lib
  4. 建立软链接
  5. [flink@dbos-bigdata-test005 lib]$ ln -s ../../../jars/hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar  hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar

image.png
7.进入平台操作安装 YARN MapReduce 框架 JAR
image.png
8.hive的辅助jar
image.png
因为后面考虑到hudi的数据要存到oss上，所以要放这几个包进来（关于oss的配置详细可参考oss配置文档）
image.png
9.重启hive，使配置生效
image.png
二、flink环境：

1.配置flink on yarn模式
配置如下：flink-conf.yaml的配置文件如下

  1. ################################################################################
  2. #  Licensed to the Apache Software Foundation (ASF) under one
  3. #  or more contributor license agreements.  See the NOTICE file
  4. #  distributed with this work for additional information
  5. #  regarding copyright ownership.  The ASF licenses this file
  6. #  to you under the Apache License, Version 2.0 (the
  7. #  "License"); you may not use this file except in compliance
  8. #  with the License.  You may obtain a copy of the License at
  9. #
  10. #      http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. #  Unless required by applicable law or agreed to in writing, software
  13. #  distributed under the License is distributed on an "AS IS" BASIS,
  14. #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. #  See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. ################################################################################
  18. #==============================================================================
  19. ## Execution
  20. ##==============================================================================
  21. # execution.target: yarn-per-job
  22. #execution.target: local
  23. execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
  24. #进行checkpointing的间隔时间（单位：毫秒）
  25. execution.checkpointing.interval: 30000
  27. execution.checkpointing.mode: EXACTLY_ONCE
  29. #execution.checkpointing.prefer-checkpoint-for-recovery: true
  30. classloader.check-leaked-classloader: false
  31. #==============================================================================
  32. # Common
  33. #==============================================================================
  35. # The external address of the host on which the JobManager runs and can be
  36. # reached by the TaskManagers and any clients which want to connect. This setting
  37. # is only used in Standalone mode and may be overwritten on the JobManager side
  38. # by specifying the --host <hostname> parameter of the bin/jobmanager.sh executable.
  39. # In high availability mode, if you use the bin/start-cluster.sh script and setup
  40. # the conf/masters file, this will be taken care of automatically. Yarn/Mesos
  41. # automatically configure the host name based on the hostname of the node where the
  42. # JobManager runs.
  44. jobmanager.rpc.address: dbos-bigdata-test005
  46. # The RPC port where the JobManager is reachable.
  48. jobmanager.rpc.port: 6123
  49. akka.framesize: 10485760b
  51. # The total process memory size for the JobManager.
  52. #
  53. # Note this accounts for all memory usage within the JobManager process, including JVM metaspace and other overhead.
  55. jobmanager.memory.process.size: 1024m
  57. # The total process memory size for the TaskManager.
  58. #
  59. # Note this accounts for all memory usage within the TaskManager process, including JVM metaspace and other overhead.
  61. #taskmanager.memory.process.size: 1728m
  62. taskmanager.heap.size: 1024m
  63. # To exclude JVM metaspace and overhead, please, use total Flink memory size instead of 'taskmanager.memory.process.size'.
  64. # It is not recommended to set both 'taskmanager.memory.process.size' and Flink memory.
  65. #
  66. # taskmanager.memory.flink.size: 1280m
  68. # The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.
  70. taskmanager.numberOfTaskSlots: 1
  72. # The parallelism used for programs that did not specify and other parallelism.
  74. parallelism.default: 1
  76. # The default file system scheme and authority.
  77. # 
  78. # By default file paths without scheme are interpreted relative to the local
  79. # root file system 'file:///'. Use this to override the default and interpret
  80. # relative paths relative to a different file system,
  81. # for example 'hdfs://mynamenode:12345'
  82. #
  83. # fs.default-scheme
  84. env.java.home key: /usr/java/jdk1.8.0_181-cloudera 
  85. #==============================================================================
  86. # High Availability
  87. #==============================================================================
  89. # The high-availability mode. Possible options are 'NONE' or 'zookeeper'.
  90. #
  91. high-availability: zookeeper
  93. # The path where metadata for master recovery is persisted. While ZooKeeper stores
  94. # the small ground truth for checkpoint and leader election, this location stores
  95. # the larger objects, like persisted dataflow graphs.
  96. # 
  97. # Must be a durable file system that is accessible from all nodes
  98. # (like HDFS, S3, Ceph, nfs, ...) 
  100. high-availability.storageDir: hdfs:///flink/ha/
  102. # The list of ZooKeeper quorum peers that coordinate the high-availability
  103. # setup. This must be a list of the form:
  104. # "host1:clientPort,host2:clientPort,..." (default clientPort: 2181)
  105. #
  106. # high-availability.zookeeper.quorum: localhost:2181
  107. high-availability.zookeeper.quorum: dbos-bigdata-test003:2181,dbos-bigdata-test004:2181,dbos-bigdata-test005:2181
  109. # ACL options are based on https://zookeeper.apache.org/doc/r3.1.2/zookeeperProgrammers.html#sc_BuiltinACLSchemes
  110. # It can be either "creator" (ZOO_CREATE_ALL_ACL) or "open" (ZOO_OPEN_ACL_UNSAFE)
  111. # The default value is "open" and it can be changed to "creator" if ZK security is enabled
  112. #
  113. # high-availability.zookeeper.client.acl: open
  115. #==============================================================================
  116. # Fault tolerance and checkpointing
  117. #==============================================================================
  119. # The backend that will be used to store operator state checkpoints if
  120. # checkpointing is enabled.
  121. #
  122. # Supported backends are 'jobmanager', 'filesystem', 'rocksdb', or the
  123. # <class-name-of-factory>.
  124. #
  125. state.backend: filesystem
  127. # Directory for checkpoints filesystem, when using any of the default bundled
  128. # state backends.
  129. #
  130. state.checkpoints.dir: hdfs://bigdata/flink-checkpoints
  131. #state.checkpoints.dir: hdfs:///flink/checkpoints
  132. #state.savepoints.dir: hdfs:///flink/savepoints
  133. # Default target directory for savepoints, optional.
  134. #
  135. # state.savepoints.dir: hdfs://namenode-host:port/flink-savepoints
  137. # Flag to enable/disable incremental checkpoints for backends that
  138. # support incremental checkpoints (like the RocksDB state backend). 
  139. #
  140. # state.backend.incremental: false
  142. # The failover strategy, i.e., how the job computation recovers from task failures.
  143. # Only restart tasks that may have been affected by the task failure, which typically includes
  144. # downstream tasks and potentially upstream tasks if their produced data is no longer available for consumption.
  146. jobmanager.execution.failover-strategy: region
  148. #==============================================================================
  149. # Rest & web frontend
  150. #==============================================================================
  152. # The port to which the REST client connects to. If rest.bind-port has
  153. # not been specified, then the server will bind to this port as well.
  154. #
  155. #rest.port: 8081
  157. # The address to which the REST client will connect to
  158. #
  159. #rest.address: 0.0.0.0
  161. # Port range for the REST and web server to bind to.
  162. #
  163. # rest.bind-port: 65535-80900
  165. # The address that the REST & web server binds to
  166. #
  167. #rest.bind-address: 0.0.0.0
  169. # Flag to specify whether job submission is enabled from the web-based
  170. # runtime monitor. Uncomment to disable.
  172. #web.submit.enable: false
  174. #==============================================================================
  175. # Advanced
  176. #==============================================================================
  178. # Override the directories for temporary files. If not specified, the
  179. # system-specific Java temporary directory (java.io.tmpdir property) is taken.
  180. #
  181. # For framework setups on Yarn or Mesos, Flink will automatically pick up the
  182. # containers' temp directories without any need for configuration.
  183. #
  184. # Add a delimited list for multiple directories, using the system directory
  185. # delimiter (colon ':' on unix) or a comma, e.g.:
  186. #     /data1/tmp:/data2/tmp:/data3/tmp
  187. #
  188. # Note: Each directory entry is read from and written to by a different I/O
  189. # thread. You can include the same directory multiple times in order to create
  190. # multiple I/O threads against that directory. This is for example relevant for
  191. # high-throughput RAIDs.
  192. #
  193. # io.tmp.dirs: /tmp
  195. # The classloading resolve order. Possible values are 'child-first' (Flink's default)
  196. # and 'parent-first' (Java's default).
  197. #
  198. # Child first classloading allows users to use different dependency/library
  199. # versions in their application than those in the classpath. Switching back
  200. # to 'parent-first' may help with debugging dependency issues.
  201. #
  202. # classloader.resolve-order: child-first
  204. # The amount of memory going to the network stack. These numbers usually need 
  205. # no tuning. Adjusting them may be necessary in case of an "Insufficient number
  206. # of network buffers" error. The default min is 64MB, the default max is 1GB.
  207. # 
  208. # taskmanager.memory.network.fraction: 0.1
  209. # taskmanager.memory.network.min: 64mb
  210. # taskmanager.memory.network.max: 1gb
  212. #==============================================================================
  213. # Flink Cluster Security Configuration
  214. #==============================================================================
  216. # Kerberos authentication for various components - Hadoop, ZooKeeper, and connectors -
  217. # may be enabled in four steps:
  218. # 1. configure the local krb5.conf file
  219. # 2. provide Kerberos credentials (either a keytab or a ticket cache w/ kinit)
  220. # 3. make the credentials available to various JAAS login contexts
  221. # 4. configure the connector to use JAAS/SASL
  223. # The below configure how Kerberos credentials are provided. A keytab will be used instead of
  224. # a ticket cache if the keytab path and principal are set.
  226. # security.kerberos.login.use-ticket-cache: true
  227. # security.kerberos.login.keytab: /path/to/kerberos/keytab
  228. # security.kerberos.login.principal: flink-user
  230. # The configuration below defines which JAAS login contexts
  232. # security.kerberos.login.contexts: Client,KafkaClient
  234. #==============================================================================
  235. # ZK Security Configuration
  236. #==============================================================================
  238. # Below configurations are applicable if ZK ensemble is configured for security
  240. # Override below configuration to provide custom ZK service name if configured
  241. # zookeeper.sasl.service-name: zookeeper
  243. # The configuration below must match one of the values set in "security.kerberos.login.contexts"
  244. # zookeeper.sasl.login-context-name: Client
  246. #==============================================================================
  247. # HistoryServer
  248. #==============================================================================
  250. # The HistoryServer is started and stopped via bin/historyserver.sh (start|stop)
  252. # Directory to upload completed jobs to. Add this directory to the list of
  253. # monitored directories of the HistoryServer as well (see below).
  254. #jobmanager.archive.fs.dir: hdfs:///completed-jobs/
  256. # The address under which the web-based HistoryServer listens.
  257. #historyserver.web.address: 0.0.0.0
  259. # The port under which the web-based HistoryServer listens.
  260. #historyserver.web.port: 8082
  262. # Comma separated list of directories to monitor for completed jobs.
  263. #historyserver.archive.fs.dir: hdfs:///completed-jobs/
  265. # Interval in milliseconds for refreshing the monitored directories.
  266. #historyserver.archive.fs.refresh-interval: 10000
  267. env.log.dir: /tmp/flink
  268. high-availability.zookeeper.path.root: /flink

2.配置flink的环境变量

  1. vim /etc/profile
  2. 以下是环境变量，根据自己的版本进行更改
  3. #set default jdk1.8 env
  4. export JAVA_HOME=/usr/java/jdk1.8.0_181-cloudera
  5. export JRE_HOME=/usr/java/jdk1.8.0_181-cloudera/jre
  6. export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
  7. export HADOOP_CONF_DIR=/etc/hadoop/conf
  8. export HADOOP_CLASSPATH=`hadoop classpath`
  9. export HBASE_CONF_DIR=/etc/hbase/conf
  10. export FLINK_HOME=/opt/flink
  11. export HIVE_HOME=/opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/hive
  12. export HIVE_CONF_DIR=/etc/hive/conf
  13. export M2_HOME=/usr/local/maven/apache-maven-3.5.4
  14. export CANAL_ADMIN_HOME=/data/canal/admin
  15. export CANAL_SERVER_HOME=/data/canal/deployer
  16. export PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin:${FLINK_HOME}/bin:${M2_HOME}/bin:${HIVE_HOME}/bin:${CANAL_SERVER_HOME}/bin:${CANAL_ADMIN_HOME}/bin:$PATH

3.查看flink是否能正常使用
image.png
4.hudi编译好的jar包和flinkcdc的jar包放到flink的lib下
image.png
flinkcdc2.1的jar包下载地址
https://github.com/ververica/flink-cdc-connectors/releases
5.以下三个包也要放到flink的lib下，否则同步数据到hive的时候会报错。
image.png
6.flink-sql的web的安装与部署

  1. 1.github上的下载地址
  2. https://github.com/zhp8341/flink-streaming-platform-web
  3. 2.安装地址
  4. https://github.com/zhp8341/flink-streaming-platform-web/blob/master/docs/deploy.md

7.编译

  1. mvn clean package  -Dmaven.test.skip=true

image.png
8.部署

  1. 2、flink-streaming-platform-web安装（一定要和flink部署在同一台）
  2. a:下载最新版本 并且解压 https://github.com/zhp8341/flink-streaming-platform-web/releases/
  3. tar -xvf   flink-streaming-platform-web.tar.gz
  4. b:执行mysql语句
  5. mysql 版本5.6+以上
  6. 创建数据库 数据库名：flink_web
  7. 执行表语句
  8. 语句地址 https://github.com/zhp8341/flink-streaming-platform-web/blob/master/docs/sql/flink_web.sql
  9. c:修改数据库连接配置
  10. /flink-streaming-platform-web/conf/application.properties  
  11. 改成上面建好的mysql地址
  12. 关于数据库连接配置 需要看清楚你 useSSL=true 你的mysql是否支持 如果不支持可以直接 useSSL=false
  13. d:启动web
  14. cd  /XXXX/flink-streaming-platform-web/bin 
  15. 启动 : sh deploy.sh  start
  16. 停止 :  sh deploy.sh  stop
  17. 日志目录地址： /XXXX/flink-streaming-platform-web/logs/
  18. 一定 一定 一定 要到bin目录下再执行deploy.sh 否则无法启动
  19. e:登录
  20. http://${ip或者hostname}:9084/  如 : http://hadoop003:9084/admin/index
  21. 登录号：admin  password: 123456
  22. -----------------------------------
  23. ©著作权归作者所有：来自51CTO博客作者江南独孤客的原创作品，请联系作者获取转载授权，否则将追究法律责任
  24. 整合Apache Hudi+Mysql+FlinkCDC2.1+CDH6.3.0
  25. https://blog.51cto.com/wang/4806267

最终的flink-web界面（支持流批一体和jar包）
image.png
三.flink cdc到hudi的demo测试

1.mysql的建表语句

  1. CREATE TABLE test_order_v4 (
  2. id int,
  3. username varchar(20),
  4. product varchar(20),
  5. price double,
  6. qty int,
  7. create_time TIMESTAMP,
  8. PRIMARY KEY (id)
  9. );

2.插入的测试数据

  1. Insert into test_order_v4 (id,username,product,price,qty,create_time) values(200,'王昱翔','芒果',12,25,current_timestamp());
  2. insert into test_order_v4 (id,username,product,price,qty,create_time) values(201,'王昱翔','芒果',12,26,current_timestamp());
  3. insert into test_order_v4 (id,username,product,price,qty,create_time) values(202,'王昱翔','芒果',12,27,current_timestamp());
  4. insert into test_order_v4 (id,username,product,price,qty,create_time) values(203,'王昱翔','芒果',12,28,current_timestamp());
  5. insert into test_order_v4 (id,username,product,price,qty,create_time) values(204,'王昱翔','芒果',12,29,current_timestamp());
  6. insert into test_order_v4 (id,username,product,price,qty,create_time) values(205,'王昱翔','芒果',12,30,current_timestamp());
  7. insert into test_order_v4 (id,username,product,price,qty,create_time) values(206,'王昱翔','芒果',12,31,current_timestamp());
  8. insert into test_order_v4 (id,username,product,price,qty,create_time) values(207,'王昱翔','芒果',12,32,current_timestamp());
  9. insert into test_order_v4 (id,username,product,price,qty,create_time) values(208,'王昱翔','芒果',12,33,current_timestamp());
  10. insert into test_order_v4 (id,username,product,price,qty,create_time) values(209,'王昱翔','芒果',12,34,current_timestamp());
  11. insert into test_order_v4 (id,username,product,price,qty,create_time) values(210,'王昱翔','芒果',12,35,current_timestamp());
  12. insert into test_order_v4 (id,username,product,price,qty,create_time) values(211,'王昱翔','芒果',12,36,current_timestamp());
  13. insert into test_order_v4 (id,username,product,price,qty,create_time) values(212,'王昱翔','芒果',12,37,current_timestamp());
  14. insert into test_order_v4 (id,username,product,price,qty,create_time) values(213,'王昱翔','芒果',12,38,current_timestamp());
  15. insert into test_order_v4 (id,username,product,price,qty,create_time) values(214,'王昱翔','芒果',12,39,current_timestamp());
  16. insert into test_order_v4 (id,username,product,price,qty,create_time) values(215,'王昱翔','芒果',12,40,current_timestamp());
  17. insert into test_order_v4 (id,username,product,price,qty,create_time) values(216,'王昱翔','芒果',12,41,current_timestamp());
  18. insert into test_order_v4 (id,username,product,price,qty,create_time) values(217,'王昱翔','芒果',12,42,current_timestamp());
  19. insert into test_order_v4 (id,username,product,price,qty,create_time) values(218,'王昱翔','芒果',12,43,current_timestamp());
  20. insert into test_order_v4 (id,username,product,price,qty,create_time) values(219,'王昱翔','芒果',12,44,current_timestamp());
  21. insert into test_order_v4 (id,username,product,price,qty,create_time) values(220,'王昱翔','芒果',12,45,current_timestamp());
  22. insert into test_order_v4 (id,username,product,price,qty,create_time) values(221,'王昱翔','芒果',12,46,current_timestamp());
  23. insert into test_order_v4 (id,username,product,price,qty,create_time) values(222,'王昱翔','芒果',12,47,current_timestamp());

3.flink-sql语句

  1. 1.创建flink cdc的表
  2. CREATE TABLE test_order_v4 (
  3. id INT,
  4. username STRING,
  5. product STRING,
  6. price DOUBLE,
  7. qty INT,
  8. create_time TIMESTAMP(0),
  9. PRIMARY KEY(id) NOT ENFORCED
  10. ) WITH (
  11. 'connector' = 'mysql-cdc',
  12. 'hostname' = '192.168.100.3',
  13. 'port' = '3306',
  14. 'username' = 'dmp',
  15. 'password' = 'wangyuxiang',
  16. 'server-time-zone' = 'Asia/Shanghai',
  17. 'debezium.snapshot.mode'='initial',
  18. 'database-name' = 'dmp',
  19. 'table-name' = 'test_order_v4'
  20. );
  21. 2.创建hudi表
  22. CREATE TABLE hudi_test_order_v4(
  23. id INT,
  24. username STRING,
  25. product STRING,
  26. price double,
  27. qty INT,
  28. create_time TIMESTAMP(0)
  29. )
  30. WITH (
  31. 'connector' = 'hudi'
  32. , 'path' = 'hdfs://bigdata/hudi/hdm2_v4'
  33. , 'hoodie.datasource.write.recordkey.field' = 'id'  -- 主键
  34. , 'write.precombine.field' = 'create_time'             -- 相同的键值时，取此字段最大值，默认ts字段
  35. , 'write.tasks' = '1'
  36. , 'compaction.tasks' = '1'
  37. , 'write.rate.limit' = '2000'                          -- 限制每秒多少条
  38. , 'table.type' = 'MERGE_ON_READ'                       -- 默认COPY_ON_WRITE
  39. , 'compaction.async.enabled' = 'true'                  -- 在线压缩
  40. , 'compaction.trigger.strategy' = 'num_commits'        -- 按次数压缩
  41. , 'compaction.delta_commits' = '5'                     -- 默认为5
  42. , 'hive_sync.enable' = 'true'                          -- 启用hive同步
  43. , 'hive_sync.mode' = 'hms'                             -- 启用hive hms同步，默认jdbc
  44. , 'hive_sync.metastore.uris' = 'thrift://dbos-bigdata-test002:9083'    -- required, metastore的端口
  45. , 'hive_sync.jdbc_url' = 'jdbc:hive2://dbos-bigdata-test002:10000'     -- required, hiveServer地址
  46. , 'hive_sync.table' = 'hudi_test_order_v4'                            -- required, hive 新建的表名
  47. , 'hive_sync.db' = 'hudi2'                              -- required, hive 新建的数据库名
  48. , 'hive_sync.username' = 'hive'                        -- required, HMS 用户名
  49. , 'hive_sync.password' = ''                            -- required, HMS Password
  50. , 'hive_sync.skip_ro_suffix' = 'true'                  -- 去除ro后缀
  51. );
  54. insert into hudi_test_order_v4 select id,username,product,price,qty,create_time from test_order_v4;

4.提交并保存
image.png
5.选择开启配置后提交任务
image.png
6.任务提交成功
image.png
7.yarn上已有这个任务
image.png
开始insert into插入数据测试
image.png
image.png
9.flink运行的DAG图上显示已插入条96数据压缩了3次
image.png

10.到hdfs上查看生成的文件
image.png
11.查看hive上是否生成表（已生成RO和RT表）
image.png
12.查询hive表中的数据

  1. select * from hudi_test_order_v4_rt;
  2. select * from hudi_test_order_v4;

说明：已自动生产hudi MOR模式的

hudi_test_order_v4(这是一个ro表，因为我在代码中去ro后缀了)

hudi_test_order_v4_rt

以下数据证明mysqlbinlog—hudi—hive的链路是成功的
image.png
四：mysql的update操作

1.先查询mysql中ID为200的数据和flink DAG目前的状态
image.png
image.png

2.mysql做update

  1. 更新一条数据
  2. UPDATE test_order_v4 set username = 'Steven'  WHERE id = 200;

image.png

  1. SELECT * from test_order_v4 WHERE id = 200;

image.png

3.更新新一条语句后，提交数从96新增加到97，但是没有进行压缩。
image.png

4.查询RT表中有此数据更新的记录

  1. select * from hudi_test_order_v4_rt;

image.png
5.查询RO表中数据没有更新。

  1. select * from hudi_test_order_v4;

image.png
五：mysql做delete的操作

1.flink DAG的状态提交97次
image.png
2.查询一条id为200的数据
image.png

3.mysql中删除此数据

  1. DELETE FROM test_order_v4 WHERE id = 200;

image.png
4.flink的DAG状态是提交了98次
image.png
6.查询hive的rt表

  1. select * from hudi_test_order_v4_rt where id = 200;
  2. hive中id=200的数据已经被删掉

image.png
7.查询hive的ro表

  1. select * from hudi_test_order_v4 where id = 200;
  2. 数据还是存在的，因为roge表没有达到触发压缩的条件，所以一直没有压缩更新

image.png
六：综合模拟频繁的更新、插入、删除测试，达到触发压缩的条件

  1. insert into test_order_v4 (id,username,product,price,qty,create_time) values(224,'王昱翔','芒果',12,47,current_timestamp());
  2. UPDATE test_order_v4 set username = 'Steven'  WHERE id = 201;
  3. UPDATE test_order_v4 set username = '王新权'  WHERE id = 202;
  4. UPDATE test_order_v4 set username = 'Steven'  WHERE id = 203;
  5. DELETE FROM test_order_v4 WHERE id = 210;
  6. DELETE FROM test_order_v4 WHERE id = 211;

1.已达到生成parquet
image.png
2.查询rt表

  1. select * from hudi_test_order_v4_rt where username = '王昱翔';

image.png
3.查询ro表

  1. select * from hudi_test_order_v4 where username = '王昱翔';

image.png
4.hudi同步到hive表中的数据做count测试

  1. select count(1) from hudi_test_order_v4 where username = '王昱翔';

image.png
image.png

  1. rt表比ro表多两条数据
  2. 总结：
  3. Hudi 表分为 COW 和 MOR两种类型
  4. COW 表适用于离线批量更新场景，对于更新数据，会先读取旧的 base file，然后合并更新数据，生成新的 base file。
  5. MOR 表适用于实时高频更新场景，更新数据会直接写入 log file 中，读时再进行合并。为了减少读放大的问题，会定期合并 log file 到 base file 中。
  8. ro表和rt表区别：
  9. ro 表全称 read oprimized table，对于 MOR 表同步的 xxx_ro 表，只暴露压缩后的 parquet。其查询方式和COW表类似。设置完 hiveInputFormat 之后 和普通的 Hive 表一样查询即可；
  10. rt表示增量视图，主要针对增量查询的rt表；
  11. ro表只能查parquet文件数据， rt表 parquet文件数据和log文件数据都可查；
  12. -----------------------------------
  13. ©著作权归作者所有：来自51CTO博客作者江南独孤客的原创作品，请联系作者获取转载授权，否则将追究法律责任
  14. 整合Apache Hudi+Mysql+FlinkCDC2.1+CDH6.3.0
  15. https://blog.51cto.com/wang/4806267