Flink-1.13集成hudi-0.10.0

把hudi的 hudi-flink-bundle_2.12-0.10.0.jar放到 flink的lib下即可从flink-sql客户端读写hudi表。

下面是完全参考hudi官网的示例

 一、下载安装flink本地集群，并将hudi的 hudi-flink-bundle_2.12-0.10.0.jar放到 flink的lib下。

1.解压

tar -zxvf flink-1.13.3-bin-scala_2.12.tgzmv flink-1.13.3-bin-scala_2.12 flink

2.配置环境变量

vi ~/.bash_profile

export Flink_HOME=/home/hadoop/flinkexport PATH=$Flink_HOME/bin:$PATHexport HADOOP_CLASSPATH=`hadoop classpath`source ~/.bash_profile

3.修改conf/flink-conf.yaml

jobmanager.rpc.address: stress102jobmanager.rpc.port: 6123jobmanager.memory.process.size: 2048mtaskmanager.memory.process.size: 8192mtaskmanager.memory.task.off-heap.size: 128mtaskmanager.numberOfTaskSlots: 32parallelism.default: 12

4.修改conf/workers增加4行数的localhost（会启动4个TaskManagerRunner）

localhostlocalhostlocalhostlocalhost

5.启动集群

export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`.bin/start-cluster.sh

启动集群之后会看到本地启动了4个和1个StandaloneSessionClusterEntrypoint

任务查看默认是在http://localhost:8081

如果8081端口没有启动要lsof -i:8081查看下是不是和其他程序冲突了,如果冲突了可以修改flink-conf.yaml的rest.port换一个端口。

https://blog.csdn.net/shy_snow/article/details/122681115

二、操作示例

1.启动flinkSql客户端 

export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

bin/sql-client.sh embedded -j ./lib/hudi-flink-bundle_2.12-0.10.0.jar

2.设格式和checkpointing

​

设置表格显示格式

set sql-client.execution.result-mode=tableau;

要设置checkpointing否则可能hudi无法写入

set execution.checkpointing.interval = 10s;

 3.创建hudi表 

​

 CREATE TABLE hudi_users
(
    id BIGINT PRIMARY KEY NOT ENFORCED,
    name STRING,
    birthday TIMESTAMP(3),
    ts TIMESTAMP(3)
)  WITH (
    'connector' = 'hudi',
    'table.type' = 'MERGE_ON_READ',
    'path' = 'hdfs://192.168.129.102:8020/hudi/hudi_users',
    'read.streaming.enabled' = 'true',
    'read.streaming.check-interval' = '3' ,
    'is_generic' = 'false'
 );

4.flinksql插入数据

insert into hudi_users values (10000000000,'DannyName',now(),now()) ;

​

该语句会提交一个任务到机器上执行

5.flinksql查询数据

select * from hudi_users;

​

如果hudi表的path上创建过不同格式的表,谨慎确认表数据可以删除之后,在创建之前从hadoop上删除掉hadoop fs -rm -r hdfs://192.168.129.102:8020/hudi/hudi_users,然后drop table hudi_user之后重新创建表。

参考: https://hudi.apache.org/docs/flink-quick-start-guide/  

YARN | Apache Flink

Overview | Apache Flink

flink集成hudi - 简书