flink集成hudi
【摘要】
Flink.png
flink 1.12.2
hudi 0.9.0
一、组件下载
1.1、flink1.12.2编译包下载：
https://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.12.2/flink-1.12.2-bin-scala_2.11.tgz
1.2、hudi编译：
git clone htt...
Flink.png
flink 1.12.2
hudi 0.9.0
一、组件下载
1.1、flink1.12.2编译包下载：
https://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.12.2/flink-1.12.2-bin-scala_2.11.tgz
1.2、hudi编译：
git clone https://github.com/apache/hudi.git && cd hudi
mvn clean package -DskipTests
注意：默认是用scala-2.11编译的
如果我们用的是flink1.12.2-2.12版本，可以自己编译成scala-2.12版本的
mvn clean package -DskipTests -Dscala-2.12
包的路径在packaging/hudi-flink-bundle/target/hudi-flink-bundle_2.12-*.*.*-SNAPSHOT.jar
建议用flink1.12.2+hudi0.9.0（master），亲测可以。
二、Batch模式具体实施步骤：
导包 hudi-flink到flink lib目录下
2.1启动flink-sql客户端，可以提前把hudi-flink-bundle_2.12-0.9.0-SNAPSHOT.jar拷贝到 $FLINK_HOME/lib目录下（我用的flink是scala2.12版本）
#HADOOP_HOME是解压二进制包后的hadoop根目录。
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
#启动flink单机集群
./bin/sql-client.sh embedded
CREATE TABLE t1(
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20)
)
PARTITIONED BY (`partition`)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://192.168.10.81:8020/hudi/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'); #查询表数据，设置一下查询模式为tableau
set execution.result-mode=tableau;
image.png
INSERT INTO t1 VALUES ('id1','Danny',24,TIMESTAMP '1970-01-01 00:00:01','par1');
image.png
三、支持stream读模式：
3.1创建表
CREATE TABLE t2(
uuid VARCHAR(20),
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20)
)
PARTITIONED BY (`partition`)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://192.168.10.81:8020/hudi/t1',
'table.type' = 'MERGE_ON_READ',
'read.streaming.enabled' = 'true', 'read.streaming.start-commit' = '20210401134557' ,
'read.streaming.check-interval' = '4'
);
这里将 table option read.streaming.enabled 设置为 true，表明通过 streaming 的方式读取表数据；
opiton read.streaming.check-interval 指定了 source 监控新的 commits 的间隔为 4s；
option table.type 设置表类型为 MERGE_ON_READ，目前只有 MERGE_ON_READ 表支持 streaming 读
image.png
insert into t1 values ('id9','test',27,TIMESTAMP '1970-01-01 00:00:01','par5');
image.png
文章来源: www.jianshu.com，作者：百忍成金的虚竹，版权归原作者所有，如需转载，请联系作者。
原文链接：www.jianshu.com/p/1c3f79916c27
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）