目錄

1、本地開發(fā)環(huán)境

2、kafka2paimon 實現(xiàn)流程

3、代碼實現(xiàn)

3.1、項目名稱

3.2、項目結構

3.3、Pom.xml 和 log4j.properties 文件

3.4、代碼核心類

3.4.1、入口類：Kafka2PaimonDemo.java

3.4.2、參數(shù)解析類

3.4.2.1、JobParameterUtil.java（ flink job scheduler ）

3.4.2.2、KafkaSourceParametersUtil.java（kafka source 參數(shù)）

3.4.2.3、OSSParametersUtils.java （oss 參數(shù)）

3.4.2.4、PaimonCatalogParameterUtils.java（paimon catalog 參數(shù)）

3.4.2.5、PaimonTableParameterUtils.java （paimon table 參數(shù)）

3.4.3、flink table env 類：FlinkTableInitUtils.java

3.4.4、配置類

3.4.4.1、DefaultConnectorVal.java

3.4.4.2、DefaultFlinkConfigVal.java

3.4.4.3、ParameterConfigs.java

3.4.4.4、PropertiesConstants.java

3.4.5、bean類

3.4.5.1、PaimonFileSystemCatalogInfo.java

3.4.5.2、JobSchedulerInfo.java

3.4.5.3、OSSGlobalVar.java

3.4.5.4、OSSInfo.java

3.4.5.5、PaimonPrimaryKeyTableSinkInfo.java

3.4.5.6、KafkaSourceInfo.java

3.4.6、重寫 OSSFileSystemFactory.java

3.5、運行核心類的步驟

3.5.1、通過本地 kafka shell 生產數(shù)據(jù)到 topic：test_paimon中

?編輯

3.5.2、編輯 main class 的 args

3.5.3、創(chuàng)建 oss bucket：flink-namespace

3.5.4、運行 Kafka2PaimonDemo.java，本地訪問 flink web-ui

4、查詢 oss 結果

4.1 paimon 表

4.2 flink checkpoint/savepoint 存儲

5、參考

---

1、本地開發(fā)環(huán)境

Mac OS 10.15.6
Oracle Jdk 11
Scala 2.12.17
Intellij Idea 2023.1
阿里云 OSS

scala 包和 jdk 包下載：

鏈接：https://pan.baidu.com/s/1HSkoUmzpbFcTx3aB9wte6w?pwd=81jc 
提取碼: 81jc

maven pom 核心依賴包：

<apache.flink.version>1.19.1</apache.flink.version>
<apache.paimon.version>0.9.0</apache.paimon.version>
<flink-kafka.version>3.3.0-1.19</flink-kafka.version>
<aliyun.oss.version>3.17.2</aliyun.oss.version>
<fs.hadoopshaded.version>3.3.0</fs.hadoopshaded.version>
<fastjson.version>1.2.83</fastjson.versi