目錄

0. 相關(guān)文章鏈接

1.?啟動(dòng)sql-client

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

1.2.?local模式

1.3.?yarn-session模式

2.?插入數(shù)據(jù)

3.?查詢數(shù)據(jù)

4.?更新數(shù)據(jù)

5.?流式插入

5.1.?創(chuàng)建測(cè)試表

5.2.?執(zhí)行插入

5.3.?查看job

5.4.?查看job

5.5. 查看HDFS目錄

5.6.?查詢結(jié)果

---

0. 相關(guān)文章鏈接

?Hudi文章匯總?

1.?啟動(dòng)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 #表示：會(huì)在本地啟動(dòng)3個(gè)TaskManager的 local集群 localhost localhost localhost

• 啟動(dòng)Flink

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

查看webui：http://bigdata1:8081

• 啟動(dòng)Flink的sql-client

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

1.3.?yarn-session模式

• 解決依賴問(wèn)題

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/

• 啟動(dòng)yarn-session

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

• 啟動(dòng)sql-client

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

2.?插入數(shù)據(jù)

set sql-client.execution.result-mode=tableau;-- 創(chuàng)建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' –- 默認(rèn)是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' );-- 插入數(shù)據(jù) 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.?查詢數(shù)據(jù)

select * from t1;

4.?更新數(shù)據(jù)

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

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

5.?流式插入

5.1.?創(chuàng)建測(cè)試表

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.?執(zhí)行插入

insert into t2 select * from sourceT;

5.3.?查看job

insert into t2 select * from sourceT;

5.4.?查看job

5.5. 查看HDFS目錄

5.6.?查詢結(jié)果

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

---

注：其他Hudi相關(guān)文章鏈接由此進(jìn) ->??Hudi文章匯總?

---

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。