ToB企服应用市场:ToB评测及商务社交产业平台

标题: 数据湖Hudi与对象存储Minio及Hive\Spark\Flink的集成 [打印本页]
作者: 千千梦丶琪    时间: 2023-2-7 20:51
标题: 数据湖Hudi与对象存储Minio及Hive\Spark\Flink的集成
    本文主要记录对象存储组件Minio、数据湖组件Hudi及查询引擎Hive\Spark之间的兼容性配置及测试情况，Spark及Hive无需多言，这里简单介绍下Minio及Hudi。
 
MinIO 是在 GNU Affero 通用公共许可证 v3.0 下发布的高性能对象存储。 它是与 Amazon S3 云存储服务兼容的 API。可使用s3a的标准接口进行读写操作。 基于 MinIO 的对象存储（Object Storage Service）服务，能够为机器学习、分析和应用程序数据工作负载构建高性能基础架构。
 
Minio官网：https://min.io/
Minio中文官网：http://www.minio.org.cn/
GitHub：https://github.com/minio/
 
Hudi 是由Uber开源的一种数据湖的存储格式，现已属于Apache顶级项目，Hudi在Hadoop文件系统之上提供了更新数据和删除数据的能力以及消费变化数据的能力。
Hudi表类型：Copy On Write
使用Parquet格式存储数据。Copy On Write表的更新操作需要通过重写实现。
Hudi表类型：Merge On Read
使用列式文件格式（Parquet）和行式文件格式（Avro）混合的方式来存储数据。Merge On Read使用列式格式存放Base数据，同时使用行式格式存放增量数据。最新写入的增量数据存放至行式文件中，根据可配置的策略执行COMPACTION操作合并增量数据至列式文件中。
Hudi官网：http://hudi.apache.org/
Hudi中文文档：http://hudi.apachecn.org/
 
 
主要的实操步骤如下：
 
一、测试环境各组件版本说明
  1. spark-3.1.2
  2. hadoop-3.2.2
  3. centos-7
  4. jdk-1.8
  5. hive-3.1.2
  6. flink-1.14.2
  7. scala-2.12.15
  8. hudi-0.11.1
  9. aws-java-sdk-1.11.563
  10. hadoop-aws-3.2.2(需要与hadoop集群版本保持一致）
复制代码
二、hive/spark的查询兼容性

 2.1、hive读取minio文件

hive-3.1.2与hadoop-3.2.2、aws的相关jar包依赖，主要分为以下部分：
  1. aws-java-sdk-1.12.363.jar
  2. aws-java-sdk-api-gateway-1.12.363.jar
  3. aws-java-sdk-bundle-1.12.363.jar
  4. aws-java-sdk-core-1.12.363.jar
  5. aws-java-sdk-s3-1.12.363.jar
  6. aws-lambda-java-core-1.2.2.jar
  7. com.amazonaws.services.s3-1.0.0.jar
  8. hadoop-aws-3.2.2.jar
复制代码
将以上jar包复制到$HIVE_HOME下；另外需要在$HADOOP_HOME/etc/hadoop下，编辑core-site.xml文件，添加以下内容：
  1. <property>
  2.   <name>fs.s3.access.key</name>
  3.   <value>minioadmin</value>
  4. </property>
  5. <property>
  6.   <name>fs.s3.secret.key</name>
  7.   <value>********</value>
  8. </property>
  9. <property>
  10.   <name>fs.s3.impl</name>
  11.   <value>org.apache.hadoop.fs.s3.S3FileSystem</value>
  12. </property>
  13. <property>
  14.   <name>fs.s3a.access.key</name>
  15.   <value>minioadmin</value>
  16. </property>
  17. <property>
  18.   <name>fs.s3a.secret.key</name>
  19.   <value>*********</value>
  20. </property>
  21. <property>
  22.   <name>fs.s3a.impl</name>
  23.   <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
  24. </property>
  25. <property>
  26.   <name>fs.s3a.endpoint</name>
  27.   <value>192.168.56.101:9000</value>
  28. </property>
  29. <property>
  30.   <name>fs.s3a.connection.ssl.enabled</name>
  31.   <value>false</value>
  32. </property>
  33. <property>
  34.   <name>fs.s3a.path.style.access</name>
  35.   <value>true</value>
  36. </property>
复制代码
 
添加完以后重启hadoop集群，并将core-site.xml分发到所有datanode，之后再复制到$HIVE_HOME/conf下。并重启hive-server2服务。
测试结果
 
在hive中新建测试表，在此之前，需要在对象存储Minio中提前上传好建表所需的文件。
如本次新上传文件为cityinfo.txt，内容为：

2.2、spark读取minio文件

hive-3.1.2与hadoop3.2.2、aws的相关jar包调试，主要分为以下部分：
  1. aws-java-sdk-1.12.363.jar
  2. aws-java-sdk-api-gateway-1.12.363.jar
  3. aws-java-sdk-bundle-1.12.363.jar
  4. aws-java-sdk-core-1.12.363.jar
  5. aws-java-sdk-s3-1.12.363.jar
  6. aws-lambda-java-core-1.2.2.jar
  7. com.amazonaws.services.s3-1.0.0.jar
  8. hadoop-aws-3.2.2.jar
复制代码
注意：此处一定要在aws官网下载目前最新的aws-java相关jar包，关于hadoop-aws-*.jar，需要与当前hadoop集群的版本适配，否则容易出现一些CLASS或PACKAGE找不到的报错。在此之前，需要将$HIVE_HOME/conf/hive-site.xml复制到$SPARK_HOME/conf下
 
测试结果
进入spark-shell客户端进行查看，命令为：
  1. $SPARK_HOME/bin/spark-shell \
  2. --conf spark.hadoop.fs.s3a.access.key=minioadmin \
  3. --conf spark.hadoop.fs.s3a.secret.key=********** \
  4. --conf spark.hadoop.fs.s3a.endpoint=192.168.56.101:9000 \
  5. --conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
  6. --master spark://192.168.56.101:7077
复制代码
注意此处命令一定要写明endpoint，需要域名或IP地址，执行以下操作：

 
可以看到无论是通过rdd方式还是通过sql方式，都可以读取到minio对象存储的文件。
2.3、dbeaver测试minio文件外部表

可通过dbeaver或其他连接工具，进行简单查询或者关联查询。

 
 
三、hudi与对象存储的互相操作可行性 

spark操作hudi

 

需要spark.hadoop.fs.s3a.xxx配置项的一些具体信息，如key\secret\endpoint等信息。
  1. spark-shell \
  2. --packages org.apache.hudi:hudi-spark3-bundle_2.11:0.11.1,org.apache.hadoop:hadoop-aws:3.2.2,com.amazonaws:aws-java-sdk:1.12.363 \
  3. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  4. --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
  5. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  6. --conf 'spark.hadoop.fs.s3a.access.key=minioadmin' \
  7. --conf 'spark.hadoop.fs.s3a.secret.key=<your-MinIO-secret-key>'\
  8. --conf 'spark.hadoop.fs.s3a.endpoint=192.168.56.101:9000' \
  9. --conf 'spark.hadoop.fs.s3a.path.style.access=true' \
  10. --conf 'fs.s3a.signing-algorithm=S3SignerType'
复制代码
以上启动spark并初始化加载hudi的jar依赖，大概输出为：
  1. hadoop@master:/opt/conf/spark/spark-3.1.2-bin-hadoop3.2/jars$ spark-shell \
  2. > --packages org.apache.hudi:hudi-spark3-bundle_2.11:0.11.1,org.apache.hadoop:hadoop-aws:3.2.2,com.amazonaws:aws-java-sdk:1.12.363 \
  3. > --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  4. > --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
  5. > --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  6. > --conf 'spark.hadoop.fs.s3a.access.key=minioadmin' \
  7. > --conf 'spark.hadoop.fs.s3a.secret.key=<your-MinIO-secret-key>'\
  8. > --conf 'spark.hadoop.fs.s3a.endpoint=192.168.56.101:9000' \
  9. > --conf 'spark.hadoop.fs.s3a.path.style.access=true' \
  10. > --conf 'fs.s3a.signing-algorithm=S3SignerType'
  11. Warning: Ignoring non-Spark config property: fs.s3a.signing-algorithm
  12. :: loading settings :: url = jar:file:/opt/conf/spark/spark-3.1.2-bin-hadoop3.2/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
  13. Ivy Default Cache set to: /home/hadoop/.ivy2/cache
  14. The jars for the packages stored in: /home/hadoop/.ivy2/jars
  15. org.apache.hudi#hudi-spark3-bundle_2.11 added as a dependency
  16. org.apache.hadoop#hadoop-aws added as a dependency
  17. com.amazonaws#aws-java-sdk added as a dependency
  18. :: resolving dependencies :: org.apache.spark#spark-submit-parent-f268e29b-59b7-4ee6-8ace-eae921495080;1.0
  19.         confs: [default]
  20.         found org.apache.hadoop#hadoop-aws;3.2.2 in central
  21.         found com.amazonaws#aws-java-sdk-bundle;1.11.563 in central
  22.         found com.amazonaws#aws-java-sdk;1.12.363 in central
  23.         found com.amazonaws#aws-java-sdk-sagemakermetrics;1.12.363 in central
  24.         found com.amazonaws#aws-java-sdk-core;1.12.363 in central
  25.         found commons-logging#commons-logging;1.1.3 in central
  26.         found commons-codec#commons-codec;1.15 in central
  27.         found org.apache.httpcomponents#httpclient;4.5.13 in central
  28.         found org.apache.httpcomponents#httpcore;4.4.13 in central
  29.         found software.amazon.ion#ion-java;1.0.2 in central
  30.         found com.fasterxml.jackson.core#jackson-databind;2.12.7.1 in central
  31.         found com.fasterxml.jackson.core#jackson-annotations;2.12.7 in central
  32.         found com.fasterxml.jackson.core#jackson-core;2.12.7 in central
  33.         found com.fasterxml.jackson.dataformat#jackson-dataformat-cbor;2.12.6 in central
  34.         found joda-time#joda-time;2.8.1 in local-m2-cache
  35.         found com.amazonaws#jmespath-java;1.12.363 in central
  36.         found com.amazonaws#aws-java-sdk-pipes;1.12.363 in central
  37.         found com.amazonaws#aws-java-sdk-sagemakergeospatial;1.12.363 in central
  38.         found com.amazonaws#aws-java-sdk-docdbelastic;1.12.363 in central
  39.         found com.amazonaws#aws-java-sdk-omics;1.12.363 in central
  40.         found com.amazonaws#aws-java-sdk-opensearchserverless;1.12.363 in central
  41.         found com.amazonaws#aws-java-sdk-securitylake;1.12.363 in central
  42.         found com.amazonaws#aws-java-sdk-simspaceweaver;1.12.363 in central
  43.         found com.amazonaws#aws-java-sdk-arczonalshift;1.12.363 in central
  44.         found com.amazonaws#aws-java-sdk-oam;1.12.363 in central
  45.         found com.amazonaws#aws-java-sdk-iotroborunner;1.12.363 in central
  46.         found com.amazonaws#aws-java-sdk-chimesdkvoice;1.12.363 in central
  47.         found com.amazonaws#aws-java-sdk-ssmsap;1.12.363 in central
  48.         found com.amazonaws#aws-java-sdk-scheduler;1.12.363 in central
  49.         found com.amazonaws#aws-java-sdk-resourceexplorer2;1.12.363 in central
  50.         found com.amazonaws#aws-java-sdk-connectcases;1.12.363 in central
  51.         found com.amazonaws#aws-java-sdk-migrationhuborchestrator;1.12.363 in central
  52.         found com.amazonaws#aws-java-sdk-iotfleetwise;1.12.363 in central
  53.         found com.amazonaws#aws-java-sdk-controltower;1.12.363 in central
  54.         found com.amazonaws#aws-java-sdk-supportapp;1.12.363 in central
  55.         found com.amazonaws#aws-java-sdk-private5g;1.12.363 in central
  56.         found com.amazonaws#aws-java-sdk-backupstorage;1.12.363 in central
  57.         found com.amazonaws#aws-java-sdk-licensemanagerusersubscriptions;1.12.363 in central
  58.         found com.amazonaws#aws-java-sdk-iamrolesanywhere;1.12.363 in central
  59.         found com.amazonaws#aws-java-sdk-redshiftserverless;1.12.363 in central
  60.         found com.amazonaws#aws-java-sdk-connectcampaign;1.12.363 in central
  61.         found com.amazonaws#aws-java-sdk-mainframemodernization;1.12.363 in central
复制代码
初次加载会稍慢，需要等候全部加载完毕。以上加载完毕后，进入改界面，证明所有依赖全部不存在冲突。如果服务器无法连接公网，则需手动安装依赖到本地仓库。

成功加载hudi部分jar包进入spark-shell客户端
之后需要依照自行编译的hudi源代码，查看hadoop、spark、hudi、scala四者之间互相依赖的jar包，大概有10数个，要注意，scala小版本之间的变化也比较频繁看，如scala-2.12.10与scala-2.12.12之间一些基础类包都有版本差异，API调用会报错，目前已针对spark-3.1.2、hadoop-3.2.3、scala-2.12.12、java8、aws-1.12.368之间的互相依赖进行了调整，能够进入spark-shell通过hudi操作minio的s3a接口，需要主要的错误如：
aws小版本与hadoop-aws协议的冲突问题，这里选定aws-java-sdk-1.12.368即可

 
以下是比较复杂的多方依赖jar包，需要通过源码的pom文件来确认小版本。
其他需要注意的点就是，某些内网环境或者屏蔽阿里云maven镜像等的服务器，需要手动安装
  1. org.apache.spark.sql.adapter.BaseSpark3Adapter
  2. org/apache/spark/internal/Logging
  3. com.fasterxml.jackson.core.jackson-annotations/2.6.0/jackson-annotations-2.6.0.jar
复制代码
  1. mvn install:install-file -Dmaven.repo.local=/data/maven/repository -DgroupId=com.fasterxml.jackson.core -DartifactId=jackson-annotations -Dversion=2.6.0 -Dpackaging=jar -Dfile=/opt/conf/spark/spark-3.1.2-bin-hadoop3.2/jars/jackson-annotations-2.6.0.jar
  3. mvn install:install-file -Dmaven.repo.local=/data/maven/repository -DgroupId=org.apache.spark -DartifactId=hudi-spark-bundle_2.12 -Dversion=0.11.1 -Dpackaging=jar -Dfile=/opt/conf/hudi/hudi-0.11.1/packaging/hudi-spark-bundle/target/original-hudi-spark-bundle_2.11-0.11.1.jar
  5. vn install:install-file -Dmaven.repo.local=/data/maven/repository -DgroupId=org.apache.hudi -DartifactId=hudi-spark-bundle_2.12 -Dversion=0.11.1 -Dpackaging=jar -Dfile=/opt/conf/hudi/hudi-0.11.1/packaging/hudi-spark-bundle/target/original-hudi-spark-bundle_2.11-0.11.1.jar
  7. mvn install:install-file -Dmaven.repo.local=/data/maven/repository -DgroupId=com.amazonaws -DartifactId=aws-java-sdk-bundle -Dversion=1.12.368 -Dpackaging=jar -Dfile=/opt/conf/spark/spark-3.1.2-bin-hadoop3.2/jars/aws-java-sdk-bundle-1.12.363.jar
复制代码
四、通过spark操作hudi进行对象存储minio的s3a接口读写

在spark集群中初始化hudi

引入相关的jar包
  1. import org.apache.hudi.QuickstartUtils._
  2. import scala.collection.JavaConversions._
  3. import org.apache.spark.sql.SaveMode._
  4. import org.apache.hudi.DataSourceReadOptions._
  5. import org.apache.hudi.DataSourceWriteOptions._
  6. import org.apache.hudi.config.HoodieWriteConfig._
  7. import org.apache.hudi.common.model.HoodieRecord
  9. val tableName = "hudi_trips_cow"
  10. val basePath = "s3a://sunshine/hudi_trips_cow"
  11. val dataGen = new DataGenerator
  13. val inserts = convertToStringList(dataGen.generateInserts(10))
  14. val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
  15. df.write.format("hudi").
  16. options(getQuickstartWriteConfigs).
  17. option("PRECOMBINE_FIELD_OPT_KEY", "ts").
  18. option("RECORDKEY_FIELD_OPT_KEY", "uuid").
  19. option("PARTITIONPATH_FIELD_OPT_KEY", "partitionpath").
  20. option("TABLE_NAME", tableName).
  21. mode(Overwrite).
  22. save(basePath)
复制代码
创建一个简单的小型 Hudi 表。Hudi DataGenerator 是一种基于示例行程模式生成示例插入和更新的快速简便的方法五、通过Flink操作hudi进行对象存储minio的s3a接口读写（待续）


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4