使用flink插入数据到hudi数据湖初探

举报
从大数据到人工智能 发表于 2022/03/28 00:29:32 2022/03/28
【摘要】 环境:hadoop 3.2.0flink 1.11.4-bin-scala_2.11hudi 0.8.0本文基于上述组件版本使用flink插入数据到hudi数据湖中。为了确保以下各步骤能够成功完成,请确保hadoop集群正常启动。确保已经配置环境变量HADOOP_CLASSPATH对于开源版本hadoop,HADOOP_CLASSPATH配置为:export HADOOP_CLASSPAT...

环境:

  • hadoop 3.2.0

  • flink 1.11.4-bin-scala_2.11

  • hudi 0.8.0

本文基于上述组件版本使用flink插入数据到hudi数据湖中。为了确保以下各步骤能够成功完成,请确保hadoop集群正常启动。

确保已经配置环境变量HADOOP_CLASSPATH

对于开源版本hadoop,HADOOP_CLASSPATH配置为:

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:$HADOOP_HOME/share/hadoop/client/*:$HADOOP_HOME/share/hadoop/common/*:$HADOOP_HOME/share/hadoop/hdfs/*:$HADOOP_HOME/share/hadoop/mapreduce/*:$HADOOP_HOME/share/hadoop/tools/*:$HADOOP_HOME/share/hadoop/yarn/*:$HADOOP_HOME/etc/hadoop/*

本文使用的hdfs为高可用集群,对应hdfs为:hdfs://mycluster

<Adsense_unit></Adsense_unit>

本地安装flink集群

  1. flink下载
wget https://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.11.4/flink-1.11.4-bin-scala_2.11.tgz
tar zxvf flink-1.11.4-bin-scala_2.11.tgz

  1. 下载hudi相关jar包,需要下载hudi-flink-bundle_2.11-0.8.0.jar、commons-logging-1.2.jar、htrace-core-3.1.0-incubating.jar以及htrace-core4-4.1.0-incubating.jar这四个jar包到flink的lib目录下,其中
cd flink-1.11.4/lib
wget https://repo.maven.apache.org/maven2/org/apache/hudi/hudi-flink-bundle_2.11/0.8.0/hudi-flink-bundle_2.11-0.8.0.jar
wget https://repo1.maven.org/maven2/commons-logging/commons-logging/1.2/commons-logging-1.2.jar
wget https://repo1.maven.org/maven2/org/apache/htrace/htrace-core/3.1.0-incubating/htrace-core-3.1.0-incubating.jar
wget https://repo1.maven.org/maven2/org/apache/htrace/htrace-core4/4.1.0-incubating/htrace-core4-4.1.0-incubating.jar
  1. 修改配置文件

  2. vi conf/workers,写入四个localhost

localhost
localhost
localhost
localhost

  1. vi conf/flink-conf.yaml,修改taskmanager.numberOfTaskSlots的值为4
taskmanager.numberOfTaskSlots: 4
  1. 启动flink集群
bin/start-cluster.sh

启动flink-sql client

  1. 执行以下命令启动flink sql
./bin/sql-client.sh embedded -j ./lib/hudi-flink-bundle_2.11-0.8.0.jar shell
  1. 创建t1表
create table t1(
 uuid VARCHAR(20),
 name VARCHAR(20),
 age INT,
 ts TIMESTAMP(3),
 `partition` VARCHAR(20)
 )
 PARTITIONED BY (`partition`)
 WITH (
   'connector'='hudi',
   'path' = 'hdfs://mycluster/tmp/t1',
   'table.type' = 'MERGE_ON_READ'
 );
  1. 插入数据到t1表
 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');
  1. 数据更新
insert into t1 values ('id1','Danny',27,TIMESTAMP '1970-01-01 00:00:01','par1');
  1. 数据查询
select * from t1 limit 10;
  1. 查询结果:

查看hdfs上对应表的分区

  1. 执行命令:
hdfs dfs -ls /tmp/t1

得到:

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。