實戰案例：Flink1.3.1 ON Hudi0.10，同步數據到Hive

由於業務需要調研數據湖的使用，這里以Hudi0.10為例，使用的是CDH6.2.1的集群。

一、編譯Hudi0.10

　　在centos7上編譯，需要配置maven，安裝scala環境和docker環境，使用集群環境為CDH6.2.1

• maven配置

  tar -zxvf apache-maven-3.6.1-bin.tar.gz -C /app
  # 配置環境變量
  export MAVEN_HOME=/app/apache-maven-3.6.1
  export PATH=${MAVEN_HOME}/bin:$PATH

  # 添加阿里雲的maven倉庫
  <mirror>
    <id>alimaven</id>
    <mirrorOf>central,!cloudera</mirrorOf>
    <name>aliyun maven</name>
    <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
  </mirror>

   

• 下載hudi0.10的源碼包進行編譯

  # 修改packging/hudi-flink-bundle的pom.xml，替換hive為2.1.1-cdh6.2.1
  
  # 編譯
  mvn clean install -DskipTests -DskipITs -Dcheckstyle.skip=true -Drat.skip=true -Dhadoop.version=3.0.0 -Pflink-bundle-shade-hive2

  

 

二、配置Flink環境（1.13.1）

• 將hudi-flink-bundle_2.11-0.10.0-SNAPSHOT.jar和hadoop-mapreduce-client*的jar放到flink1.13.1的lib目錄下

  mv ./hudi-flink-bundle_2.11-0.10.0.jar /app/flink-1.13.1/lib
  
  cd /app/flink-1.13.1/lib
  cd /opt/cloudera/parcels/CDH/jars/hadoop-mapreduce-client-common-3.0.0-cdh6.2.1.jar ./
  cd /opt/cloudera/parcels/CDH/jars/hadoop-mapreduce-client-core-3.0.0-cdh6.2.1.jar ./
  cd /opt/cloudera/parcels/CDH/jars/hadoop-mapreduce-client-jobclient-3.0.0-cdh6.2.1 ./

• 配置Flink On Yarn模式

  # flink_conf.yaml
  
  execution.target: yarn-per-job
  execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
  execution.checkpointing.interval: 30000
  execution.checkpointing.mode: EXACTLY_ONCE
  classloader.check-leaked-classloader: false
  
  jobmanager.rpc.address: hadoop001
  jobmanager.rpc.port: 6123
  jobmanager.memory.process.size: 1600m
  taskmanager.memory.process.size: 1728m
  taskmanager.numberOfTaskSlots: 1
  parallelism.default: 1
  state.backend: filesystem
  state.checkpoints.dir: hdfs://hadoop001:8020/flink-checkpoints
  jobmanager.execution.failover-strategy: region

   

• 配置Flink，Hadoop，Hive，HBase的環境變量

  
  export JAVA_HOME=/usr/java/jdk1.8.0_231
  export JRE_HOME=${JAVA_HOME}/jre
  export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
  export PATH=${JAVA_HOME}/bin:$PATH
  
  export HADOOP_HOME=/opt/cloudera/parcels/CDH/lib/hadoop
  export HADOOP_CONF_DIR=/etc/hadoop/conf
  export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
  export FLINK_HOME=/app/flink-1.13.1
  export PATH=${FLINK_HOME}/bin:$PATH
  export HIVE_HOME=/opt/cloudera/parcels/CDH/lib/HIVE
  export HIVE_CONF_DIR=/etc/hive/conf
  export HBASE_CONF_DIR=/etc/hbase/conf

   

三、部署同步hive環境

• 將hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar放到../CDH/jars 和 ../CDH/lib/hive/lib下面，每個節點都需要

  cp hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar /opt/cloudera/parcels/CDH/jars/
  
  cd ../lib/hive/lib
  ln -ls ../../../jars/hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar

• 安裝YARN-MapReduce的jar

  

•  設置hive輔助jar

   

   將hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar放到hive輔助jar路徑下面， 上述步驟后需要重啟hive meta和server2服務

• cp hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar /usr/local/src/hook/hive　　　

 

四、測試用例

　　1）測試Kafka數據往Hudi寫，並且同步到Hive

• 創建Kafka Topic

  cd /opt/cloudera/parcels/CDH/lib/kafka/bin
  
  ./kafka-topics.sh --zookeeper hadoop001:2181,hadoop002:2181,hadoop003:2181 --create --topic hudi_sync --replication-factor 1 --partitions 1

   

• 啟動flink sql client

  cd /app/flink
  ./bin/sql-client.sh embedded

   

• 創建source，sink表，執行插入操作

  # kafka source 表， 需要將kafka-connector放到flink lib下
  CREATE TABLE t_source (
     id STRING
    ,name STRING
    ,age INT
    ,create_time STRING
    ,par STRING
  ) WITH (
      'connector' = 'kafka',  -- 使用 kafka connector
      'topic' = 'hudi_sync',  -- kafka topic名稱
      'scan.startup.mode' = 'earliest-offset',  -- 從起始 offset 開始讀取
      'properties.bootstrap.servers' = 'hadoop001:9092,hadoop002:9092,hadoop003:9092',  -- kafka broker 地址
      'properties.group.id' = 'group2', 
      'value.format' = 'json',
      'value.json.fail-on-missing-field' = 'true',
      'value.fields-include' = 'ALL'
  );
  
  # hudi表：這里創建的是COW表，適用於離線批量
  CREATE TABLE t_hdm(
      id VARCHAR(20)
    ,name VARCHAR(30)
    ,age INT
    ,create_time VARCHAR(30)
    ,par VARCHAR(20)
   )
  PARTITIONED BY (par) 
   WITH (
       'connector' = 'hudi'
     , 'path' = 'hdfs://hadoop001/hudi/hdm6'
     , 'hoodie.datasource.write.recordkey.field' = 'id'  -- 主鍵
     , 'write.precombine.field' = 'age'             -- 相同的鍵值時，取此字段最大值，默認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://hadoop001:9083'    -- required, metastore的端口
     , 'hive_sync.jdbc_url' = 'jdbc:hive2://hadoop001:10000'     -- required, hiveServer地址
     , 'hive_sync.table' = 'hdm'                            -- required, hive 新建的表名
     , 'hive_sync.db' = 'hudi'                              -- required, hive 新建的數據庫名
     , 'hive_sync.username' = ''                        -- required, HMS 用戶名
     , 'hive_sync.password' = ''                            -- required, HMS 密碼
     , 'hive_sync.skip_ro_suffix' = 'true'                  -- 去除ro后綴
   );
  
  
  -- 寫入數據
  insert into t_hdm
  select id, name, age, create_time, par from t_source;

   

• 測試數據

  {"id": "id1", "name": "Danny", "age": 23, "create_time": "1970-01-01 00:00:01", "par": "par1"}
  {"id": "id2", "name": "Danny1", "age": 24, "create_time": "1970-01-01 00:00:07", "par": "par1"}
  {"id": "id3", "name": "Danny2", "age": 25, "create_time": "1970-01-01 00:01:01", "par": "par2"}
  {"id": "id4", "name": "Danny3", "age": 26, "create_time": "1970-01-01 00:02:08", "par": "par2"}
  {"id": "id5", "name": "Danny5", "age": 28, "create_time": "1970-01-01 00:04:12", "par": "par4"}

   

• hudi中存儲為parquet

  

   

• hive beeline查詢，記得設置input format： 

  set hive.input.format = org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat;

  

   

　　2）MySQL CDC 入湖

• 將CDC jar放到Flink lib目錄下

  cd /app/flink
  mv /opt/softwares/flink-sql-connector-mysql-cdc-2.1.0.jar ./
  mv /opt/softwares/flink-format-changelog-json-2.1.0.jar ./

   

• SQL Client提交任務

  -- mysql source
  CREATE TABLE mysql_users (
      userId STRING PRIMARY KEY NOT ENFORCED ,
      userName STRING
  ) WITH (
      'connector'= 'mysql-cdc',
      'hostname'= 'node',
      'port'= '3306',
      'username'= 'root',
      'password'= '123456',
      'server-time-zone'= 'Asia/Shanghai',
      'debezium.snapshot.mode' = 'initial',
      'database-name'= 'aucc',
      'table-name'= 'dim_user'
  );
  
  -- 創建臨時視圖, 主要為了添加part字段，用於hive分區
  create view user_view AS 
  SELECT *, DATE_FORMAT(now(), 'yyyyMMdd') as part
  FROM mysql_users;
  
  -- hudi sink
  CREATE TABLE t_cdc_hdm(
      userId STRING,
      userName STRING,
      par VARCHAR(20),
      primary key(userId) not enforced
  )
  PARTITIONED BY (par)
  with(
      'connector'='hudi',
      'path'= 'hdfs://hadoop001/hudi/hdm8'
      , 'hoodie.datasource.write.recordkey.field'= 'userId'-- 主鍵
      , 'write.precombine.field'= 'ts'-- 自動precombine的字段
      , 'write.tasks'= '1'
      , 'compaction.tasks'= '1'
      , 'write.rate.limit'= '2000'-- 限速
      , 'table.type'= 'MERGE_ON_READ'-- 默認COPY_ON_WRITE,可選MERGE_ON_READ 
      , 'compaction.async.enabled'= 'true'-- 是否開啟異步壓縮
      , 'compaction.trigger.strategy'= 'num_commits'-- 按次數壓縮
      , 'compaction.delta_commits'= '1'-- 默認為5
      , 'changelog.enabled'= 'true'-- 開啟changelog變更
      , 'read.streaming.enabled'= 'true'-- 開啟流讀
      , 'read.streaming.check-interval'= '3'-- 檢查間隔，默認60s
      , 'hive_sync.enable'= 'true'-- 開啟自動同步hive
      , 'hive_sync.mode'= 'hms'-- 自動同步hive模式，默認jdbc模式
      , 'hive_sync.metastore.uris'= 'thrift://hadoop001:9083'-- hive metastore地址
      -- , 'hive_sync.jdbc_url'= 'jdbc:hive2://hadoop:10000'-- hiveServer地址
      , 'hive_sync.table'= 't_mysql_cdc'-- hive 新建表名
      , 'hive_sync.db'= 'hudi'-- hive 新建數據庫名
      , 'hive_sync.username'= ''-- HMS 用戶名
      , 'hive_sync.password'= ''-- HMS 密碼
      , 'hive_sync.support_timestamp'= 'true'-- 兼容hive timestamp類型
  );
  
  insert into t_cdc_hdm select userId, userName, part as par  from user_view;