目錄

0. 相關文章鏈接

1.?啟動sql-client

1.1.?修改flink-conf.yaml配置

1.2.?local模式

1.3.?yarn-session模式

2.?插入數據

3.?查詢數據

4.?更新數據

5.?流式插入

5.1.?創建測試表

5.2.?執行插入

5.3.?查看job

5.4.?查看job

5.5. 查看HDFS目錄

5.6.?查詢結果

---

0. 相關文章鏈接

?Hudi文章匯總?

1.?啟動sql-client

1.1.?修改flink-conf.yaml配置

vim /opt/module/flink-1.13.6/conf/flink-conf.yamlclassloader.check-leaked-classloader: false taskmanager.numberOfTaskSlots: 4state.backend: rocksdb execution.checkpointing.interval: 30000 state.checkpoints.dir: hdfs://hadoop1:8020/ckps state.backend.incremental: true

1.2.?local模式

• 修改workers

vim /opt/module/flink-1.13.6/conf/workers #表示：會在本地啟動3個TaskManager的 local集群 localhost localhost localhost

• 啟動Flink

/opt/module/flink-1.13.6/bin/start-cluster.sh

查看webui：http://bigdata1:8081

• 啟動Flink的sql-client

/opt/module/flink-1.13.6/bin/sql-client.sh embedded

1.3.?yarn-session模式

• 解決依賴問題

cp /opt/module/hadoop-3.1.3/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.1.3.jar /opt/module/flink-1.13.6/lib/

• 啟動yarn-session

/opt/module/flink-1.13.6/bin/yarn-session.sh -d

• 啟動sql-client

/opt/module/flink-1.13.6/bin/sql-client.sh embedded -s yarn-session

2.?插入數據

set sql-client.execution.result-mode=tableau;-- 創建hudi表 CREATE TABLE t1(uuid VARCHAR(20) PRIMARY KEY NOT ENFORCED,name VARCHAR(10),age INT,ts TIMESTAMP(3),`partition` VARCHAR(20) ) PARTITIONED BY (`partition`) WITH ('connector' = 'hudi','path' = 'hdfs://hadoop1:8020/tmp/hudi_flink/t1','table.type' = 'MERGE_ON_READ' –- 默認是COW ); -- 或如下寫法 CREATE TABLE t1(uuid VARCHAR(20),name VARCHAR(10),age INT,ts TIMESTAMP(3),`partition` VARCHAR(20),PRIMARY KEY(uuid) NOT ENFORCED ) PARTITIONED BY (`partition`) WITH ('connector' = 'hudi','path' = 'hdfs://hadoop1:8020/tmp/hudi_flink/t1','table.type' = 'MERGE_ON_READ' );-- 插入數據 INSERT INTO t1 VALUES('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'),('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');

3.?查詢數據

select * from t1;

4.?更新數據

insert into t1 values ('id1','Danny',27,TIMESTAMP '1970-01-01 00:00:01','par1');

????????注意，保存模式現在是Append。通常，除非是第一次創建表，否則請始終使用追加模式?，F在再次查詢數據將顯示更新的記錄。每個寫操作都會生成一個用時間戳表示的新提交。查找前一次提交中相同的_hoodie_record_keys在_hoodie_commit_time、age字段中的變化。

5.?流式插入

5.1.?創建測試表

CREATE TABLE sourceT (uuid varchar(20),name varchar(10),age int,ts timestamp(3),`partition` varchar(20) ) WITH ('connector' = 'datagen','rows-per-second' = '1' );create table t2(uuid varchar(20),name varchar(10),age int,ts timestamp(3),`partition` varchar(20) ) with ('connector' = 'hudi','path' = '/tmp/hudi_flink/t2','table.type' = 'MERGE_ON_READ' );

5.2.?執行插入

insert into t2 select * from sourceT;

5.3.?查看job

insert into t2 select * from sourceT;

5.4.?查看job

5.5. 查看HDFS目錄

5.6.?查詢結果

set sql-client.execution.result-mode=tableau; select * from t2 limit 10;

---

注：其他Hudi相關文章鏈接由此進 ->??Hudi文章匯總?

---

總結

以上是生活随笔為你收集整理的Hudi（12）：Hudi集成Flink之sql-client方式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。