前言
筆者我集群使用的Flink1.12.5，Hudi官方版本是不支持Flink1.12.3-Flink1.12.5之間的Flink版本，所以我對Hudi原始碼進行了一部分修改，最終可以實作Flink1.12.5與Hudi0.9正常使用（包括sync hive功能），理論上來說 Flink 1.12.3-Flink1.12.5之間都可以通過這套方法來進行Flink與Hudi的適配，
筆者的環境：Hadoop3.1.1、hive3.1.0 、Flink 1.12.5、Hudi0.9.0
Hudi 0.9的原始碼地址：https://github.com/apache/hudi/tree/release-0.9.0
以下是修改程序
-
修改hudi-release-0.9目錄下的pom.xml檔案
①、注釋掉hudi-integ-test 與packaging/hudi-integ-test-bundle不注釋掉這兩個打包的時候會報錯
②、將 <flink.version>1.12.2</flink.version> 改為
? <flink.version>1.12.5</flink.version>
-
修改 hudi-release-0.9.0/hudi-flink/src/main/java\org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
①、添加屬性 private transient SubtaskGateway[] gateways;
②、實作 subtaskReady 方法，并在方法內部添加操作
@Override
public void subtaskReady(int subtask, SubtaskGateway gateway) {
this.gateways[subtask] = gateway;
}
③、實作 notifyCheckpointAborted 方法，不做任何操作
④、修改sendCommitAckEvents()方法的回傳值與捕獲的例外型別如下：
⑤、洗掉importorg.apache.flink.runtime.operators.coordination.TaskNotRunningException; 不洗掉在編譯程序中會報錯
⑥、在 start() 方法內添加
this.gateways = new SubtaskGateway[this.parallelism];
-
在 hudi-release-0.9.0/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java 內在 StreamWriteFunctionWrapper(String tablePath, Configuration conf) 方法內，洗掉 MockOperatorCoordinatorContext弟三個引數,修改后如下：
-
修改hudi-release-0.9.0/packaging/hudi-flink-bundle/pom.xml檔案
1. 修改下面的屬性，換成適配你hive的 flink-sql-connector-hive 依賴，這個依賴maven沒有，需要去Flink官網 https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/hive/ 下載后手動添加到本地maven倉庫中，正確添加后第二條屬性在idea里面會檢查到，不報紅，下面兩條屬性用的版本號要一致， <include>org.apache.flink:flink-sql-connector-hive-2.3.6_${scala.binary.version}</include> <profile> <id>include-flink-sql-connector-hive</id> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-sql-connector-hive-2.3.6_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> </dependencies> </profile>下面是我修改后的適配hive 3.1.0版本的兩條屬性
兩條屬性的位置是不一樣的
-
在hudi-release-0.9.0目錄下，執行maven打包命令
hive 3 系列版本（默認打包hive 3.1.2的依賴）: mvn install -DskipTests -Drat.skip=true -Pflink-bundle-shade-hive3
hive 2 系列版本（默認打包hive 2.3.6的依賴）: mvn install -DskipTests -Drat.skip=true -Pflink-bundle-shade-hive2
hive 1系列版本（默認打包hive1.1.0的依賴） mvn install -DskipTests -Drat.skip=true -Pflink-bundle-shade-hive1
如果版本默認打包的hive版本號與你使用的版本號不兼容，則修改 hudi-release-0.9.0/packaging/hudi-flink-bundle/pom.xml 內 對應 flink-bundle-shade-hive 屬性的具體的hive版本號
<profile> <id>flink-bundle-shade-hive1</id> <properties> <hive.version>1.1.0</hive.version> <thrift.version>0.9.2</thrift.version> <flink.bundle.hive.scope>compile</flink.bundle.hive.scope> </properties> </profile> <profile> <id>flink-bundle-shade-hive2</id> <properties> <hive.version>2.3.1</hive.version> <flink.bundle.hive.scope>compile</flink.bundle.hive.scope> </properties> <dependencies> <dependency> <groupId>${hive.groupid}</groupId> <artifactId>hive-service-rpc</artifactId> <version>${hive.version}</version> <scope>${flink.bundle.hive.scope}</scope> </dependency> </dependencies> </profile> <profile> <id>flink-bundle-shade-hive3</id> <properties> <hive.version>3.1.2</hive.version> <flink.bundle.hive.scope>compile</flink.bundle.hive.scope> </properties> <dependencies> <dependency> <groupId>${hive.groupid}</groupId> <artifactId>hive-service-rpc</artifactId> <version>${hive.version}</version> <scope>${flink.bundle.hive.scope}</scope> </dependency> </dependencies> </profile> -
編譯完成后，可以在hudi-release-0.9.0\packaging\hudi-flink-bundle\target下找到編譯好的 hudi-flink-bundle_2.11-0.9.0.jar 包，可以在idea里外部引入這個包或者放入到Flink的lib目錄下使之生效
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/298901.html
標籤：其他
上一篇：你們想知道的一切，都在這里了。
下一篇：Linux網路配置