Zeppelin結合Flink查詢hudi資料

關於Zeppelin

Zeppelin是基於 Web 的notebook，是支援使用 SQL、Scala、Python、R 等進行資料驅動的互動式資料分析和協作的Notebook。

Zeppelin支援多種語言後端，Apache Zeppelin 直譯器允許將任何語言/資料處理後端插入 Zeppelin。目前 Apache Zeppelin 支援 Apache Spark、Apache Flink、Python、R、JDBC、Markdown 和 Shell 等多種直譯器。

（關於Flink和Hudi介紹，可參考本博主其他文章，或 search by yourself）

下面今日正題。

本文涉及元件及其版本

元件名稱

版本號

hadoop

3。2。0

hudi

0。10。0-SNAPSHOT

zeppelin

0。10。0

flink

1。13。1

在執行以下操作之前，請先將資料匯入hudi，如果還未匯入，可參考：

https：//lrting。top/flink/flink-sql-client-savepoint-example。html

相關部落格文章將資料匯入hudi

zeppelin安裝包下載

mkdir /data && cd /datawget https：//dlcdn。apache。org/zeppelin/zeppelin-0。10。0/zeppelin-0。10。0-bin-all。tgztar zxvf zeppelin-0。10。0-bin-all。tgzln -s /data/zeppelin-0。10。0-bin-all /data/zeppelin

zeppelin配置檔案修改

cd /data/zeppelin/confcp zeppelin-site。xml。template zeppelin-site。xml

將zeppelin。server。addr配置項修改為0。0。0。0

zeppelin預設埠使用8080，如果和你本地埠衝突可將其改為別的埠，本文件將埠修改為8008，也就是將zeppelin。server。port配置項修改為8008

cp zeppelin-env。sh。template zeppelin-env。sh

填入以下變數：

export JAVA_HOME=/data/jdkexport HADOOP_CONF_DIR=/data/hadoop/etc/hadoopexport FLINK_HOME=/data/flink

各變數請根據自己環境進行設定。

本文後續使用預設的local模式啟動Flink。

啟動zeppelin

bin/zeppelin-daemon。sh start

此時如果你們沒有建立logs資料夾以及run資料夾，那麼會在zeppelin目錄下自動建立，如下所示：

［root@hadoop zeppelin］# bin/zeppelin-daemon。sh startLog dir doesn‘t exist， create /data/zeppelin/logsPid dir doesn’t exist， create /data/zeppelin/runZeppelin start ［ OK ］

此時瀏覽器輸入zeppelin伺服器ip：8008或者hostname：8008即可進入如下頁面：

基本使用

點選Notebook，點選Create new note，填寫文字名稱以及選定flink interpreter，如下所示：

新建完之後進入如下頁面：

根據前面說到的，我們已經透過文章

https：//lrting。top/flink/flink-sql-client-savepoint-example。html

所述將資料匯入hudi，那麼此時我們就可以進行查詢：

我們選擇

%flink。ssql

首先定義hudi表：

create table stu8_binlog_sink_hudi（ id bigint not null， name string， `school` string， nickname string， age int not null， score decimal（4，2） not null， class_num int not null， phone bigint not null， email string， ip string， primary key （id） not enforced） partitioned by （`school`） with （ ‘connector’ = ‘hudi’， ‘path’ = ‘hdfs：//hadoop：9000/tmp/test_stu8_binlog_sink_hudi’， ‘table。type’ = ‘MERGE_ON_READ’， ‘write。precombine。field’ = ‘school’ ）；

對hudi表進行統計：

select * from stu8_binlog_sink_hudi；

得到如下結果：