浙江龙网

  1. 首页 > 生活百科 > >

Hudi 数据湖的插入，更新，查询，分析操作示例

生活百科

Hudi 数据湖的插入 ， 更新，查询，分析操作示例【Hudi 数据湖的插入，更新，查询，分析操作示例】作者：Grey
原文地址：
博客园：Hudi 数据湖的插入，更新，查询 ， 分析操作示例
CSDN：Hudi 数据湖的插入 ， 更新，查询，分析操作示例
前置工作首先，需要先完成
Linux 下搭建 Kafka 环境
Linux 下搭建 Hadoop 环境
Linux 下搭建 HBase 环境
Linux 下搭建 Hive 环境
本文基于上述四个环境已经搭建完成的基础上进行 Hudi 数据湖的插入，更新，查询操作 。
开发环境Scala 2.11.8
JDK 1.8
需要熟悉 Maven 构建项目和 Scala 一些基础语法 。
操作步骤master 节点首先启动集群，执行：
stop-dfs.sh && start-dfs.sh启动 yarn，执行：
stop-yarn.sh && start-yarn.sh然后准备一个 Mave 项目 ， 在 src/main/resources 目录下 ， 将 Hadoop 的一些配置文件拷贝进来，分别是
$HADOOP_HOME/etc/hadoop/core-site.xml 文件
<?xml version="1.0" encoding="UTF-8"?><?xml-stylesheet type="text/xsl" href="https://www.huyubaike.com/biancheng/configuration.xsl"?><configuration><property><name>fs.default.name</name><value>hdfs://master:9000</value></property><property><name>hadoop.tmp.dir</name><value>/usr/local/hadoop/tmp</value></property></configuration>注意，需要在你访问集群的机器上配置 host 文件，这样才可以识别 master 节点 。
$HADOOP_HOME/etc/hadoop/hdfs-site.xml 文件
<?xml version="1.0" encoding="UTF-8"?><?xml-stylesheet type="text/xsl" href="https://www.huyubaike.com/biancheng/configuration.xsl"?><configuration><property><name>dfs.replication</name><value>1</value></property><property><name>dfs.permissions</name><value>false</value></property></configuration>$HADOOP_HOME/etc/hadoop/yarn-site.xml 文件，目前还没有任何配置
<?xml version="1.0"?><configuration></configuration>然后，设计实体的数据结构，
package git.snippet.entitycase class MyEntity(uid: Int,uname: String,dt: String)插入数据代码如下
package git.snippet.testimport git.snippet.entity.MyEntityimport git.snippet.util.JsonUtilimport org.apache.spark.SparkConfimport org.apache.spark.sql.{SaveMode, SparkSession}object DataInsertion {def main(args: Array[String]): Unit = {System.setProperty("HADOOP_USER_NAME", "root")val sparkConf = new SparkConf().setAppName("MyFirstDataApp").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer").setMaster("local[*]")val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()val ssc = sparkSession.sparkContextssc.hadoopConfiguration.set("dfs.client.use.datanode.hostname", "true")insertData(sparkSession)}def insertData(sparkSession: SparkSession) = {import org.apache.spark.sql.functions._import sparkSession.implicits._val commitTime = System.currentTimeMillis().toString //生成提交时间val df = sparkSession.read.text("/mydata/data1").mapPartitions(partitions => {partitions.map(item => {val jsonObject = JsonUtil.getJsonData(item.getString(0))MyEntity(jsonObject.getIntValue("uid"), jsonObject.getString("uname"), jsonObject.getString("dt"))})})val result = df.withColumn("ts", lit(commitTime)) //添加ts 时间戳列.withColumn("uuid", col("uid")).withColumn("hudipart", col("dt")) //增加hudi分区列result.write.format("org.apache.hudi").option("hoodie.insert.shuffle.parallelism", 2).option("hoodie.upsert.shuffle.parallelism", 2).option("PRECOMBINE_FIELD_OPT_KEY", "ts") //指定提交时间列.option("RECORDKEY_FIELD_OPT_KEY", "uuid") //指定uuid唯一标示列.option("hoodie.table.name", "myDataTable").option("hoodie.datasource.write.partitionpath.field", "hudipart") //分区列.mode(SaveMode.Overwrite).save("/snippet/data/hudi")}}然后，在 master 节点先准备好数据
vi data1输入如下数据
{'uid':1,'uname':'grey','dt':'2022/09'}{'uid':2,'uname':'tony','dt':'2022/10'}然后创建文件目录，
hdfs dfs -mkdir /mydata/把 data1 放入目录下
hdfs dfs -put data1 /mydata/访问：http://192.168.100.130:50070/explorer.html#/mydata
可以查到这个数据

Hudi 数据湖的插入，更新，查询，分析操作示例

文章插图
接下来执行插入数据的 scala 代码 ， 执行完毕后，验证一下
访问：http://192.168.100.130:50070/explorer.html#/snippet/data/hudi/2022
可以查看到插入的数据

推荐阅读


上一篇：Autobus 方法记录

下一篇：day01-4-订座功能