Hudi 集成 | 超详细步骤！整合 Apache Hudi + Flink + CDH

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

mvn clean install -DskipTests -DskipITs -Dcheckstyle.skip=true -Drat.skip=true -Dhadoop.version=3.0.0  -Pflink-bundle-shade-hive2

execution.target: yarn-per-job#execution.target: localexecution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION#进行checkpointing的间隔时间（单位：毫秒）execution.checkpointing.interval: 30000 execution.checkpointing.mode: EXACTLY_ONCE #execution.checkpointing.prefer-checkpoint-for-recovery: trueclassloader.check-leaked-classloader: false jobmanager.rpc.address: dbos-bigdata-test005 # The RPC port where the JobManager is reachable. jobmanager.rpc.port: 6123akka.framesize: 10485760b jobmanager.memory.process.size: 1024m taskmanager.heap.size: 1024m taskmanager.numberOfTaskSlots: 1 # The parallelism used for programs that did not specify and other parallelism. parallelism.default: 1 env.java.home key: /usr/java/jdk1.8.0_181-cloudera  high-availability: zookeeper high-availability.storageDir: hdfs:///flink/ha/ high-availability.zookeeper.quorum: dbos-bigdata-test003:2181,dbos-bigdata-test004:2181,dbos-bigdata-test005:2181 state.backend: filesystem # Directory for checkpoints filesystem, when using any of the default bundled# state backends.#state.checkpoints.dir: hdfs://bigdata/flink-checkpoints jobmanager.execution.failover-strategy: region env.log.dir: /tmp/flinkhigh-availability.zookeeper.path.root: /flink

vim /etc/profile以下是环境变量，根据自己的版本进行更改#set default jdk1.8 envexport JAVA_HOME=/usr/java/jdk1.8.0_181-clouderaexport JRE_HOME=/usr/java/jdk1.8.0_181-cloudera/jreexport CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/libexport HADOOP_CONF_DIR=/etc/hadoop/confexport HADOOP_CLASSPATH=`hadoop classpath`export HBASE_CONF_DIR=/etc/hbase/confexport FLINK_HOME=/opt/flinkexport HIVE_HOME=/opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/hiveexport HIVE_CONF_DIR=/etc/hive/confexport M2_HOME=/usr/local/maven/apache-maven-3.5.4export CANAL_ADMIN_HOME=/data/canal/adminexport CANAL_SERVER_HOME=/data/canal/deployerexport 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

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

[flink@dbos-bigdata-test005 lib]$ pwd/opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/hive/lib//建立软链接[flink@dbos-bigdata-test005 lib]$ ln -ls ../../../jars/hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar  hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar

//创建topickafka-topics --zookeeper  dbos-bigdata-test003:2181,dbos-bigdata-test004:2181,dbos-bigdata-test005:2181/kafka --create --partitions 4 --replication-factor 3 --topic test  //删除topickafka-topics --zookeeper  dbos-bigdata-test003:2181,dbos-bigdata-test004:2181,dbos-bigdata-test005:2181/kafka --delete --topic test//生产数据kafka-console-producer --broker-list dbos-bigdata-test003:9092,dbos-bigdata-test004:9092,dbos-bigdata-test005:9092 --topic test//直接复制数据{"tinyint0": 6, "smallint1": 223, "int2": 42999, "bigint3": 429450, "float4": 95.47324181659323, "double5": 340.5755392968011,"decimal6": 111.1111, "boolean7": true,  "char8": "dddddd", "varchar9": "buy0", "string10": "buy1", "timestamp11": "2021-09-13 03:08:50.810"}

[flink@dbos-bigdata-test005 hive]$ cd  /opt/flink[flink@dbos-bigdata-test005 flink]$ lltotal 496drwxrwxr-x  2 flink flink   4096 May 25 20:36 bindrwxrwxr-x  2 flink flink   4096 Nov  4 17:22 confdrwxrwxr-x  7 flink flink   4096 May 25 20:36 examplesdrwxrwxr-x  2 flink flink   4096 Nov  4 13:58 lib-rw-r--r--  1 flink flink  11357 Oct 29  2019 LICENSEdrwxrwxr-x  2 flink flink   4096 May 25 20:37 licensesdrwxr-xr-x  2 flink flink   4096 Jan 30  2021 log-rw-rw-r--  1 flink flink 455180 May 25 20:37 NOTICEdrwxrwxr-x  3 flink flink   4096 May 25 20:36 optdrwxrwxr-x 10 flink flink   4096 May 25 20:36 plugins-rw-r--r--  1 flink flink   1309 Jan 30  2021 README.txt[flink@dbos-bigdata-test005 flink]$ ./bin/sql-client.sh

Hudi 表分为 COW 和 MOR两种类型COW 表适用于离线批量更新场景，对于更新数据，会先读取旧的 base file，然后合并更新数据，生成新的 base file。MOR 表适用于实时高频更新场景，更新数据会直接写入 log file 中，读时再进行合并。为了减少读放大的问题，会定期合并 log file 到 base file 中。

//创建source表CREATE TABLE k (   tinyint0 TINYINT  ,smallint1 SMALLINT  ,int2 INT  ,bigint3 BIGINT  ,float4 FLOAT  ,double5 DOUBLE    ,decimal6 DECIMAL(38,8)  ,boolean7 BOOLEAN  ,char8 STRING  ,varchar9 STRING  ,string10 STRING  ,timestamp11 STRING) WITH (    'connector' = 'kafka',  -- 使用 kafka connector    'topic' = 'test',  -- kafka topic名称    'scan.startup.mode' = 'earliest-offset',  -- 从起始 offset 开始读取    'properties.bootstrap.servers' = 'dbos-bigdata-test003:9092,dbos-bigdata-test005:9092,dbos-bigdata-test005:9092',  -- kafka broker 地址    'properties.group.id' = 'testgroup1',     'value.format' = 'json',    'value.json.fail-on-missing-field' = 'true',    'value.fields-include' = 'ALL');

// 创建Hudi（cow）sink表CREATE TABLE hdm(   tinyint0 TINYINT  ,smallint1 SMALLINT  ,int2 INT  ,bigint3 BIGINT  ,float4 FLOAT  ,double5 DOUBLE    ,decimal6 DECIMAL(12,3)  ,boolean7 BOOLEAN  ,char8 CHAR(64)  ,varchar9 VARCHAR(64)  ,string10 STRING  ,timestamp11 TIMESTAMP(3) )PARTITIONED BY (tinyint0)  WITH (     'connector' = 'hudi'   , 'path' = 'hdfs://bigdata/hudi/hdm'   , 'hoodie.datasource.write.recordkey.field' = 'char8'  -- 主键   , 'write.precombine.field' = 'timestamp11'             -- 相同的键值时，取此字段最大值，默认ts字段   , 'write.tasks' = '1'   , 'compaction.tasks' = '1'   , 'write.rate.limit' = '2000'                          -- 限制每秒多少条   , 'compaction.async.enabled' = 'true'                  -- 在线压缩   , 'compaction.trigger.strategy' = 'num_commits'        -- 按次数压缩   , 'compaction.delta_commits' = '5'                     -- 默认为5   , 'hive_sync.enable' = 'true'                          -- 启用hive同步   , 'hive_sync.mode' = 'hms'                             -- 启用hive hms同步，默认jdbc   , 'hive_sync.metastore.uris' = 'thrift://dbos-bigdata-test002:9083'    -- required, metastore的端口   , 'hive_sync.jdbc_url' = 'jdbc:hive2://dbos-bigdata-test002:10000'     -- required, hiveServer地址   , 'hive_sync.table' = 'hdm'                            -- required, hive 新建的表名   , 'hive_sync.db' = 'hudi'                              -- required, hive 新建的数据库名   , 'hive_sync.username' = 'hive'                        -- required, HMS 用户名   , 'hive_sync.password' = ''                            -- required, HMS 密码   , 'hive_sync.skip_ro_suffix' = 'true'                  -- 去除ro后缀 );// 创建Hudi（mor）sink表CREATE TABLE hdm(   tinyint0 TINYINT  ,smallint1 SMALLINT  ,int2 INT  ,bigint3 BIGINT  ,float4 FLOAT  ,double5 DOUBLE    ,decimal6 DECIMAL(12,3)  ,boolean7 BOOLEAN  ,char8 CHAR(64)  ,varchar9 VARCHAR(64)  ,string10 STRING  ,timestamp11 TIMESTAMP(3) )PARTITIONED BY (tinyint0)  WITH (     'connector' = 'hudi'   , 'path' = 'hdfs://bigdata/hudi/hdm'   , 'hoodie.datasource.write.recordkey.field' = 'char8'  -- 主键   , 'write.precombine.field' = 'timestamp11'             -- 相同的键值时，取此字段最大值，默认ts字段   , 'write.tasks' = '1'   , 'compaction.tasks' = '1'   , 'write.rate.limit' = '2000'                          -- 限制每秒多少条   , 'table.type' = 'MERGE_ON_READ'                       -- 默认COPY_ON_WRITE   , 'compaction.async.enabled' = 'true'                  -- 在线压缩   , 'compaction.trigger.strategy' = 'num_commits'        -- 按次数压缩   , 'compaction.delta_commits' = '5'                     -- 默认为5   , 'hive_sync.enable' = 'true'                          -- 启用hive同步   , 'hive_sync.mode' = 'hms'                             -- 启用hive hms同步，默认jdbc   , 'hive_sync.metastore.uris' = 'thrift://dbos-bigdata-test002:9083'    -- required, metastore的端口   , 'hive_sync.jdbc_url' = 'jdbc:hive2://dbos-bigdata-test002:10000'     -- required, hiveServer地址   , 'hive_sync.table' = 'hdm'                            -- required, hive 新建的表名   , 'hive_sync.db' = 'hudi'                              -- required, hive 新建的数据库名   , 'hive_sync.username' = 'hive'                        -- required, HMS 用户名   , 'hive_sync.password' = ''                            -- required, HMS 密码   , 'hive_sync.skip_ro_suffix' = 'true'                  -- 去除ro后缀 );

// 插入source数据 insert into hdm select         cast(tinyint0 as TINYINT)    , cast(smallint1 as SMALLINT)    , cast(int2 as INT)    , cast(bigint3 as BIGINT)    , cast(float4 as FLOAT)    , cast(double5 as DOUBLE)    , cast(decimal6 as DECIMAL(38,18))    , cast(boolean7 as BOOLEAN)    , cast(char8 as CHAR(64))    , cast(varchar9 as VARCHAR(64))    , cast(string10 as STRING)    , cast(timestamp11 as TIMESTAMP(3))  from  k;

2021-11-04 16:17:29,687 ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Could not start cluster entrypoint YarnJobClusterEntrypoint.org.apache.flink.runtime.entrypoint.ClusterEntrypointException: Failed to initialize the cluster entrypoint YarnJobClusterEntrypoint.  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:212) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:600) [flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.yarn.entrypoint.YarnJobClusterEntrypoint.main(YarnJobClusterEntrypoint.java:99) [flink-dist_2.11-1.13.1.jar:1.13.1] Caused by: org.apache.flink.util.FlinkException: Could not create the DispatcherResourceManagerComponent.  at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:275) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:250) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$1(ClusterEntrypoint.java:189) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_181]  at javax.security.auth.Subject.doAs(Subject.java:422) ~[?:1.8.0_181]  at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]  at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:186) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  ... 2 more Caused by: java.net.BindException: Could not start rest endpoint on any port in port range 40631  at org.apache.flink.runtime.rest.RestServerEndpoint.start(RestServerEndpoint.java:234) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory.create(DefaultDispatcherResourceManagerComponentFactory.java:172) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runCluster(ClusterEntrypoint.java:250) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.lambda$startCluster$1(ClusterEntrypoint.java:189) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_181]  at javax.security.auth.Subject.doAs(Subject.java:422) ~[?:1.8.0_181]  at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1875) ~[hadoop-common-3.0.0-cdh6.3.0.jar:?]  at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:186) ~[flink-dist_2.11-1.13.1.jar:1.13.1]  ... 2 more