CDH集群下，Flink+hive+iceberg+zeppelin实践—01

偏执的太偏执、 2023-01-15 11:28 134阅读 0赞

# 实操记录 #

### 版本： ###

Flink 1.11.2

iceberg 最新版本

zeppelin

hive 是基于cdh 6.3.2版本下的hive 2.1

注意，iceberg只支持2.x以上，官网说的。

## 一，保证Flink on zeppelin 查询hive没问题 ##

1）Zeppelin flink 参数配置：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

*  Flink sql 读写hive

1）Jar包放入Flink lib下：

flink-connector-hive\_2.11-1.11.2.jar

hive-exec-2.1.0.jar

2）可能需要导入hadoop包  
          flink-hadoop-compatibility\_2.11-1.11.2.jar

flink-shaded-hadoop\-2\-uber\-2.6.0\-cdh5.16.2-9.0.jar（这要自己编译）  
      flink-shaded-hadoop-2-uber-2.7.5-9.0.jar（实际测试使用这个）

3) 配置环境变量

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

实际配置为：

> export HADOOP\_CLASSPATH=\`hadoop classpath\`
> 
> export HADOOP\_CONF\_DIR=/etc/hadoop/conf.cloudera.hdfs
> 
> export HADOOP\_HOME=/wyyt/software/cloudera/parcels/CDH/lib/hadoop
> 
> export PATH=$HADOOP\_HOME/bin:$HADOOP\_HOME/sbin:$PATH
> 
>  
> 
> export HIVE\_CONF\_DIR=/etc/hive/conf
> 
> export HIVE\_HOME=/wyyt/software/cloudera/parcels/CDH/lib/hive
> 
> export PATH=$HIVE\_HOME/bin:$HIVE\_HOME/sbin:$PATH

看一下Flink lib下文件：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

测试：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

## 二，在zeppelin操作读写iceberg ##

### 1，导入依赖 ###

![20210422172352295.png][]

### 2，创建catalog ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

hadoop catalog：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

### 3，查看catalog是否成功 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6][]

### 4，我们创建一个database ###

CREATE DATABASE iceberg\_db;

使用hive iceberg\_db库；

use iceberg\_db;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7][]

### 5，创建iceberg表 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8][]

### 6，插入数据 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9][]

任务插入完成：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10][]

### 7，查询数据 ###

官网说查询可以使用2种方式：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11][]

实际的zeppelin使用这里会报错：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12][]

***正确的方式：***

***batch方式：***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13][]

**streaming方式，一直运行：**

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14][]

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15][]

### 8，查看数据情况 ###

在hdfs上面查看数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16][]

# 9，代码操作 #

### 1，代码 ###

>     public class Flink2IcebergByHiveCatalog {
>     
>         public static void main(String[] args) throws Exception{
>     
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>             env.setParallelism(1);
>             env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
>             EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
>             StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);
>             bsTableEnv.executeSql("show catalogs").print();
>     
>     
>             bsTableEnv.executeSql("CREATE CATALOG hive_catalog2  WITH (\n" +
>                     "'type'='iceberg',\n" +
>                     "'catalog-type'='hive',\n" +
>                     "'uri'='thrift://192.168.5.xx:9083',\n" +
>                     "'clients'='5',\n" +
>                     "'property-version'='1',\n" +
>                     "'warehouse'='hdfs://192.168.5.xx:8020/tmp'\n" +
>                     ")");
>     
>             bsTableEnv.executeSql("show catalogs").print();
>             bsTableEnv.executeSql("use catalog hive_catalog2");
>             bsTableEnv.executeSql("use  iceberg_db");
>             bsTableEnv.executeSql("show  tables").print();
>     
>             bsTableEnv.executeSql("CREATE TABLE hive_catalog2.iceberg_db.sample3 (\n" +
>                     "    id BIGINT COMMENT 'unique id',\n" +
>                     "    data STRING\n" +
>                     ")");
>     
>             //todo 插入数据
>             bsTableEnv.executeSql("INSERT INTO hive_catalog2.iceberg_db.sample3 VALUES (1, 'c')");
>             bsTableEnv.executeSql("SELECT * FROM sample3").print();
>     
>     
>     
>         }
>     }

### 2，依赖 ###

>     <properties>
>           <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
>           <flink.version>1.11.2</flink.version>
>           <scala.binary.version>2.11</scala.binary.version>
>           <scala.version>2.11.12</scala.version>
>       </properties>
>     
>     
>       <dependencies>
>     
>     
>           
>           <dependency>
>               <groupId>org.apache.iceberg</groupId>
>               <artifactId>iceberg-flink-runtime</artifactId>
>               <version>0.11.1</version>
>               <scope>provided</scope>
>           </dependency>
>     
>           <dependency>
>               <groupId>org.apache.hadoop</groupId>
>               <artifactId>hadoop-client</artifactId>
>               <version>2.7.3</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>org.apache.hadoop</groupId>
>               <artifactId>hadoop-common</artifactId>
>               <version>2.7.3</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>joda-time</groupId>
>               <artifactId>joda-time</artifactId>
>               <version>2.10.5</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-connector-kafka_2.11</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>           
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-json</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-core</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-table-common</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>     
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-clients_${scala.binary.version}</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>     
>           </dependency>
>     
>     
>           
>           <dependency>
>               <groupId>org.apache.flink</groupId>
>               <artifactId>flink-connector-hive_2.11</artifactId>
>               <version>${flink.version}</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>           
>           <dependency>
>               <groupId>com.alibaba</groupId>
>               <artifactId>fastjson</artifactId>
>               <version>1.2.60</version>
>               
>           </dependency>
>           <dependency>
>               <groupId>org.apache.hive</groupId>
>               <artifactId>hive-exec</artifactId>
>               <version>2.1.0</version>
>               <scope>provided</scope>
>           </dependency>
>     
>     
>       </dependencies>

### 10，iceberg文件解析： ###

参考：Iceberg文件解析（[https://blog.csdn.net/weixin\_47482194/article/details/115676338?spm=1001.2014.3001.5501][https_blog.csdn.net_weixin_47482194_article_details_115676338_spm_1001.2014.3001.5501]）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221022/80705e3ed0634190a9517bd1121189af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221022/822fb73ba9904de7b382d4219e90514e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221022/bb4645b672b94196b3dbbee256f9df64.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221022/6c3d5442c4a4410b9595893771d781c7.png
[20210422172352295.png]: /images/20221022/3a497d6f01c64693ba194d8bf0bcb251.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221022/2f5b953ba9f34500a839d2c7abc4350b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221022/022d801a0b1542ac88cd8139e1a34e98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6]: /images/20221022/da270f378b2a4c5684f466328438d670.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7]: /images/20221022/6653b6c7a5cd46f8b4357b62659c687d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8]: /images/20221022/8de24bc00dc74403afd0976c33e57043.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9]: /images/20221022/558038dd277241b7a4fcf5d95637bdf1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10]: /images/20221022/9fa0eb17a9cc4918a2a71a14da60ef19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11]: /images/20221022/70b941d2f51e4020ae324e729dec50e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12]: /images/20221022/849c46abb999432da697d514d1d80070.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13]: /images/20221022/4d35b1be88c34b07bc31daa6cd92496d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14]: /images/20221022/7879db0f77b644e499990380c927d016.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15]: /images/20221022/1939b19e57e34ab384195d7ffeff34a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16]: /images/20221022/dc18f7a22ebc40dda13b5ddebf85864e.png
[https_blog.csdn.net_weixin_47482194_article_details_115676338_spm_1001.2014.3001.5501]: https://blog.csdn.net/weixin_47482194/article/details/115676338?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17]: /images/20221022/8a31872914234cbdb9dd76a99a146aa2.png