0
0

数据湖之Hudi（9）：使用Spark向Hudi中插入数据

0. 相关文章链接

大数据基础知识点 文章汇总

1. 开发说明

Apache Hudi最初是由Uber开发的，旨在以高效率实现低延迟的数据库访问。Hudi 提供了Hudi 表的概念，这些表支持CRUD操作，基于Spark框架使用Hudi API 进行读写操作。

2. 环境构建

2.1. 构建服务器环境

关于构建Spark向Hudi中插入数据的服务器环境，可以参考博文的另外一篇博文，在CentOS7上安装HDFS即可，博文连接：数据湖之Hudi（6）：Hudi与Spark和HDFS的集成安装使用

2.2. 构建Maven项目

需要在IDEA中创建一个Maven工程，并将服务器上的core-site.xml 和 hdfs-site.xml 这2个配置文件导入，以及创建一个log4j.properties文件，如下图所示：

log4j.properties 文件内容如下：

  1. log4j.rootCategory=WARN, console
  2. log4j.rootLogger=error,stdout
  3. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
  4. log4j.appender.stdout.target=System.out
  5. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
  6. log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

注意，这是本地跑程序，需要配置好域名映射。

3. Maven依赖

  1.     <repositories>
  2.         <repository>
  3.             <id>aliyun</id>
  4.             <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
  5.         </repository>
  6.         <repository>
  7.             <id>cloudera</id>
  8.             <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
  9.         </repository>
  10.         <repository>
  11.             <id>jboss</id>
  12.             <url>http://repository.jboss.com/nexus/content/groups/public</url>
  13.         </repository>
  14.     </repositories>
  16.     <properties>
  17.         <scala.version>2.12.10</scala.version>
  18.         <scala.binary.version>2.12</scala.binary.version>
  19.         <spark.version>3.0.0</spark.version>
  20.         <hadoop.version>3.0.0</hadoop.version>
  21.         <hudi.version>0.9.0</hudi.version>
  22.     </properties>
  24.     <dependencies>
  26.         <!-- 依赖Scala语言 -->
  27.         <dependency>
  28.             <groupId>org.scala-lang</groupId>
  29.             <artifactId>scala-library</artifactId>
  30.             <version>${scala.version}</version>
  31.         </dependency>
  33.         <!-- Spark Core 依赖 -->
  34.         <dependency>
  35.             <groupId>org.apache.spark</groupId>
  36.             <artifactId>spark-core_${scala.binary.version}</artifactId>
  37.             <version>${spark.version}</version>
  38.         </dependency>
  39.         <!-- Spark SQL 依赖 -->
  40.         <dependency>
  41.             <groupId>org.apache.spark</groupId>
  42.             <artifactId>spark-sql_${scala.binary.version}</artifactId>
  43.             <version>${spark.version}</version>
  44.         </dependency>
  46.         <!-- Hadoop Client 依赖 -->
  47.         <dependency>
  48.             <groupId>org.apache.hadoop</groupId>
  49.             <artifactId>hadoop-client</artifactId>
  50.             <version>${hadoop.version}</version>
  51.         </dependency>
  53.         <!-- hudi-spark3 -->
  54.         <dependency>
  55.             <groupId>org.apache.hudi</groupId>
  56.             <artifactId>hudi-spark3-bundle_2.12</artifactId>
  57.             <version>${hudi.version}</version>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>org.apache.spark</groupId>
  61.             <artifactId>spark-avro_2.12</artifactId>
  62.             <version>${spark.version}</version>
  63.         </dependency>
  65.     </dependencies>
  67.     <build>
  68.         <outputDirectory>target/classes</outputDirectory>
  69.         <testOutputDirectory>target/test-classes</testOutputDirectory>
  70.         <resources>
  71.             <resource>
  72.                 <directory>${project.basedir}/src/main/resources</directory>
  73.             </resource>
  74.         </resources>
  75.         <!-- Maven 编译的插件 -->
  76.         <plugins>
  77.             <plugin>
  78.                 <groupId>org.apache.maven.plugins</groupId>
  79.                 <artifactId>maven-compiler-plugin</artifactId>
  80.                 <version>3.0</version>
  81.                 <configuration>
  82.                     <source>1.8</source>
  83.                     <target>1.8</target>
  84.                     <encoding>UTF-8</encoding>
  85.                 </configuration>
  86.             </plugin>
  87.             <plugin>
  88.                 <groupId>net.alchim31.maven</groupId>
  89.                 <artifactId>scala-maven-plugin</artifactId>
  90.                 <version>3.2.0</version>
  91.                 <executions>
  92.                     <execution>
  93.                         <goals>
  94.                             <goal>compile</goal>
  95.                             <goal>testCompile</goal>
  96.                         </goals>
  97.                     </execution>
  98.                 </executions>
  99.             </plugin>
  100.         </plugins>
  101.     </build>

4. 核心代码

在上述图片的包中新建scala的object对象，对象名为：Demo01_InsertForCOW，用于实现模拟数据，插入Hudi表，采用COW模式。

具体需求：使用官方QuickstartUtils提供模拟产生Trip数据，模拟100条交易Trip乘车数据，将其转换为DataFrame数据集，保存至Hudi表中，代码基本与spark-shell命令行一致

具体代码如下：

  1. package com.ouyang.hudi.crud
  3. import org.apache.hudi.QuickstartUtils.DataGenerator
  4. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
  6. /**
  7.  * @ date: 2022/2/23
  8.  * @ author: yangshibiao
  9.  * @ desc: 模拟数据，插入Hudi表，采用COW模式
  10.  * 使用官方QuickstartUtils提供模拟产生Trip数据，
  11.  * 模拟100条交易Trip乘车数据，将其转换为DataFrame数据集，
  12.  * 保存至Hudi表中，代码基本与spark-shell命令行一致
  13.  */
  14. object Demo01_InsertForCOW {
  16.     def main(args: Array[String]): Unit = {
  18.         // 创建SparkSession实例对象，设置属性
  19.         val spark: SparkSession = {
  20.             SparkSession.builder()
  21.                 .appName(this.getClass.getSimpleName.stripSuffix("$"))
  22.                 .master("local[4]")
  23.                 // 设置序列化方式：Kryo
  24.                 .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  25.                 .getOrCreate()
  26.         }
  28.         // 定义变量：表名称、保存路径
  29.         val tableName: String = "tbl_trips_cow"
  30.         val tablePath: String = "/hudi-warehouse/tbl_trips_cow"
  32.         // 构建数据生成器，模拟产生业务数据
  33.         import org.apache.hudi.QuickstartUtils._
  34.         import scala.collection.JavaConverters._
  35.         import spark.implicits._
  37.         // 第1步、模拟乘车数据
  38.         val dataGen: DataGenerator = new DataGenerator()
  39.         val inserts = convertToStringList(dataGen.generateInserts(100))
  41.         // 将集合对象写入到df中
  42.         val insertDF: DataFrame = spark.read.json(
  43.             spark.sparkContext.parallelize(inserts.asScala, 2).toDS()
  44.         )
  45.         insertDF.printSchema()
  46.         insertDF.show(10, truncate = false)
  48.         // TOOD: 第2步、插入数据到Hudi表
  49.         import org.apache.hudi.DataSourceWriteOptions._
  50.         import org.apache.hudi.config.HoodieWriteConfig._
  51.         insertDF.write
  52.             .mode(SaveMode.Append)
  53.             .format("hudi")
  54.             .option("hoodie.insert.shuffle.parallelism", "2")
  55.             .option("hoodie.upsert.shuffle.parallelism", "2")
  56.             // Hudi 表的属性值设置
  57.             .option(PRECOMBINE_FIELD.key(), "ts")
  58.             .option(RECORDKEY_FIELD.key(), "uuid")
  59.             .option(PARTITIONPATH_FIELD.key(), "partitionpath")
  60.             .option(TBL_NAME.key(), tableName)
  61.             .save(tablePath)
  62.     }
  63. }

点击执行后可能会碰到 null\bin\winutils.exe in the Hadoop binaries 问题，这个是在windows本地执行时没有对应环境，可以忽略，如下图所示：

在代码中打印了数据格式和部分数据，如下所示：

  1. root
  2.  |-- begin_lat: double (nullable = true)
  3.  |-- begin_lon: double (nullable = true)
  4.  |-- driver: string (nullable = true)
  5.  |-- end_lat: double (nullable = true)
  6.  |-- end_lon: double (nullable = true)
  7.  |-- fare: double (nullable = true)
  8.  |-- partitionpath: string (nullable = true)
  9.  |-- rider: string (nullable = true)
  10.  |-- ts: long (nullable = true)
  11.  |-- uuid: string (nullable = true)
  13. +-------------------+-------------------+----------+-------------------+-------------------+------------------+------------------------------------+---------+-------------+------------------------------------+
  14. |begin_lat          |begin_lon          |driver    |end_lat            |end_lon            |fare              |partitionpath                       |rider    |ts           |uuid                                |
  15. +-------------------+-------------------+----------+-------------------+-------------------+------------------+------------------------------------+---------+-------------+------------------------------------+
  16. |0.4726905879569653 |0.46157858450465483|driver-213|0.754803407008858  |0.9671159942018241 |34.158284716382845|americas/brazil/sao_paulo           |rider-213|1645620263263|550e7186-203c-48a8-9964-edf12e0dfbe3|
  17. |0.6100070562136587 |0.8779402295427752 |driver-213|0.3407870505929602 |0.5030798142293655 |43.4923811219014  |americas/brazil/sao_paulo           |rider-213|1645074858260|c8d5e237-6589-419e-bef7-221faa4faa13|
  18. |0.5731835407930634 |0.4923479652912024 |driver-213|0.08988581780930216|0.42520899698713666|64.27696295884016 |americas/united_states/san_francisco|rider-213|1645298902122|d64b94ec-d8e8-44f3-a5c0-e205e034aa5d|
  19. |0.21624150367601136|0.14285051259466197|driver-213|0.5890949624813784 |0.0966823831927115 |93.56018115236618 |americas/united_states/san_francisco|rider-213|1645132033863|fd8f9051-b5d2-4403-8002-8bb173df5dc8|
  20. |0.40613510977307   |0.5644092139040959 |driver-213|0.798706304941517  |0.02698359227182834|17.851135255091155|asia/india/chennai                  |rider-213|1645254343160|160c7699-7f5e-4ec3-ba76-9ae63ae815af|
  21. |0.8742041526408587 |0.7528268153249502 |driver-213|0.9197827128888302 |0.362464770874404  |19.179139106643607|americas/united_states/san_francisco|rider-213|1645452263906|fe9d75c0-f326-4cef-8596-4248a57d1fea|
  22. |0.1856488085068272 |0.9694586417848392 |driver-213|0.38186367037201974|0.25252652214479043|33.92216483948643 |americas/united_states/san_francisco|rider-213|1645133755620|5d149bc7-78a8-46df-b2b0-a038dc79e378|
  23. |0.0750588760043035 |0.03844104444445928|driver-213|0.04376353354538354|0.6346040067610669 |66.62084366450246 |americas/brazil/sao_paulo           |rider-213|1645362187498|da2dd8e5-c2d9-45e2-8c96-520927e5458d|
  24. |0.651058505660742  |0.8192868687714224 |driver-213|0.20714896002914462|0.06224031095826987|41.06290929046368 |asia/india/chennai                  |rider-213|1645575914370|f01e9d28-df30-454c-a780-b56cd5b43ce7|
  25. |0.11488393157088261|0.6273212202489661 |driver-213|0.7454678537511295 |0.3954939864908973 |27.79478688582596 |americas/united_states/san_francisco|rider-213|1645094601577|bd4ae628-3885-4b26-8a50-c14f8e42a265|
  26. +-------------------+-------------------+----------+-------------------+-------------------+------------------+------------------------------------+---------+-------------+------------------------------------+
  27. only showing top 10 rows

运行程序后会发现数据已经插入到HDFS中了，如下图所示：

注：Hudi系列博文为通过对Hudi官网学习记录所写，其中有加入个人理解，如有不足，请各位读者谅解☺☺☺

注：****其他相关文章链接由此进（包括Hudi在内的各大数据相关博文） -> 大数据基础知识点 文章汇总

标签： spark 大数据 big data
本文转载自: https://blog.csdn.net/yang_shibiao/article/details/123100943
版权归原作者 电光闪烁 所有， 如有侵权，请联系我们删除。
登录后发布评论

“数据湖之Hudi（9）：使用Spark向Hudi中插入数据”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？