揭秘数据湖长文详解Hudi从内核到实战（一）

　　Hudi入门与构建
　　Hudi介绍
　　Hudi将带来流式处理大数据，提供新数据集，同时比传统批处理效率高一个数据量级。
　　Hudi主要特性：  快速upsert,可插入索引；  以原子方式操作数据并具有回滚功能；  写入器之间的快照隔离；  savepoint用户数据恢复的保存点；  管理文件大小，使用统计数据布局；  数据行的异步压缩和柱状数据；  时间轴数据跟踪血统。
　　Hudi快速构建  安装环境准备  Hadoop集群  Hive  Spark2.4.5(2.x)  Maven安装
　　把apache-maven-3.6.1-bin.tar.gz上传到linux的/opt/software目录下。
　　解压apache-maven-3.6.1-bin.tar.gz到/opt/module/目录下面。  [atguigu@hadoop102 software]$ tar -zxvf apache-maven-3.6.1-bin.tar.gz -C /opt/module/
　　修改apache-maven-3.6.1的名称为maven。  [atguigu@hadoop102 module]$ mv apache-maven-3.6.1/ maven
　　添加环境变量到/etc/profile中。  [atguigu@hadoop102 module]$ sudo vim /etc/profile #MAVEN_HOME export MAVEN_HOME=/opt/module/maven export PATH=$PATH:$MAVEN_HOME/bin
　　测试安装结果。  [atguigu@hadoop102 module]$ source /etc/profile [atguigu@hadoop102 module]$ mvn -v
　　修改setting.xml，指定为阿里云。  [atguigu@hadoop102 maven]$ cd conf [atguigu@hadoop102 maven]$ vim settings.xml           nexus-aliyun        central         Nexusaliyun        http://maven.aliyun.com/nexus/content/groups/public 
　　Git安装  [atguigu@hadoop102 software]$ sudo yum install git [atguigu@hadoop102 software]$ git --version构建Hudi  [atguigu@hadoop102 software]$ cd /opt/module/ [atguigu@hadoop102 module]$ git clone https://github.com/apache/hudi.git && cd hudi [atguigu@hadoop102 hudi]$ vim pom.xml                 nexus-aliyun         nexus-aliyun        http://maven.aliyun.com/nexus/content/groups/public/                      true                              false               [atguigu@hadoop102hudi]$ mvn clean package -DskipTests -DskipITs
　　通过Spark-shell快速开始  Spark-shell启动
　　spark-shell启动,需要指定spark-avro模块，因为默认环境里没有，spark-avro模块版本好需要和spark版本对应，这里都是2.4.5。  [root@hadoop102hudi]# spark-shell --packages org.apache.spark:spark-avro_2.11:2.4.5  --conf＂spark.serializer=org.apache.spark.serializer.KryoSerializer＂  --jars /opt/module/hudi/packaging/hudi-spark-bundle/target/hudi-spark-bundle_2.11-0.6.0-SNAPSHOT.jar设置表名
　　设置表名，基本路径和数据生成器。  scala>import org.apache.hudi.QuickstartUtils._ import org.apache.hudi.QuickstartUtils._   scala>import scala.collection.JavaConversions._ import scala.collection.JavaConversions._   scala>import org.apache.spark.sql.SaveMode._ import org.apache.spark.sql.SaveMode._   scala>import org.apache.hudi.DataSourceReadOptions._ import org.apache.hudi.DataSourceReadOptions._   scala>import org.apache.hudi.DataSourceWriteOptions._ import org.apache.hudi.DataSourceWriteOptions._   scala>import org.apache.hudi.config.HoodieWriteConfig._ import org.apache.hudi.config.HoodieWriteConfig._   scala>val tableName = ＂hudi_trips_cow＂ tableName:String = hudi_trips_cow   scala>val basePath = ＂file:///tmp/hudi_trips_cow＂ basePath:String = file:///tmp/hudi_trips_cow   scala>val dataGen = new DataGenerator dataGen:org.apache.hudi.QuickstartUtils.DataGenerator =org.apache.hudi.QuickstartUtils$DataGenerator@5cdd5ff9插入数据
　　新增数据，生成一些数据，将其加载到DataFrame中，然后将DataFrame写入Hudi表。  scala>val inserts = convertToStringList(dataGen.generateInserts(10)) scala>val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2)) scala>df.write.format(＂hudi＂).         options(getQuickstartWriteConfigs).         option(PRECOMBINE_FIELD_OPT_KEY,＂ts＂).         option(RECORDKEY_FIELD_OPT_KEY,＂uuid＂).         option(PARTITIONPATH_FIELD_OPT_KEY,＂partitionpath＂).         option(TABLE_NAME, tableName).         mode(Overwrite).         save(basePath)
　　Mode(overwrite)将覆盖重新创建表（如果已存在）。可以检查/tmp/hudi_trps_cow路径下是否有数据生成。  [root@hadoop102~]# cd /tmp/hudi_trips_cow/ [root@hadoop102hudi_trips_cow]# ls americas  asia查询数据  scala>val tripsSnapshotDF = spark.      |  read.     |   format(＂hudi＂).      |  load(basePath + ＂/*/*/*/*＂) scala>tripsSnapshotDF.createOrReplaceTempView(＂hudi_trips_snapshot＂) scala>spark.sql(＂select fare, begin_lon, begin_lat, ts from  hudi_trips_snapshot where fare >20.0＂).show() +------------------+-------------------+-------------------+---+ |              fare|          begin_lon|          begin_lat| ts| +------------------+-------------------+-------------------+---+ |64.27696295884016| 0.4923479652912024| 0.5731835407930634|0.0| |33.92216483948643| 0.9694586417848392| 0.1856488085068272|0.0| |27.79478688582596| 0.6273212202489661|0.11488393157088261|0.0| |93.56018115236618|0.14285051259466197|0.21624150367601136|0.0| |  43.4923811219014| 0.8779402295427752|0.6100070562136587|0.0| |66.62084366450246|0.03844104444445928| 0.0750588760043035|0.0| |34.158284716382845|0.46157858450465483|0.4726905879569653|0.0| |41.06290929046368| 0.8192868687714224| 0.651058505660742|0.0| +------------------+-------------------+-------------------+---+ scala>spark.sql(＂select _hoodie_commit_time, _hoodie_record_key,_hoodie_partition_path, rider, driver, fare from  hudi_trips_snapshot＂).show() +-------------------+--------------------+----------------------+---------+----------+------------------+ |_hoodie_commit_time| _hoodie_record_key|_hoodie_partition_path|    rider|   driver|              fare| +-------------------+--------------------+----------------------+---------+----------+------------------+ |     20200701105144|6007a624-d942-4e0...|  americas/united_s...|rider-213|driver-213|64.27696295884016| |     20200701105144|db7c6361-3f05-48d...|  americas/united_s...|rider-213|driver-213|33.92216483948643| |     20200701105144|dfd0e7d9-f10c-468...| americas/united_s...|rider-213|driver-213|19.179139106643607| |     20200701105144|e36365c8-5b3a-415...|  americas/united_s...|rider-213|driver-213|27.79478688582596| |     20200701105144|fb92c00e-dea2-48e...|  americas/united_s...|rider-213|driver-213|93.56018115236618| |     20200701105144|98be3080-a058-47d...| americas/brazil/s...|rider-213|driver-213|  43.4923811219014| |     20200701105144|3dd6ef72-4196-469...|  americas/brazil/s...|rider-213|driver-213|66.62084366450246| |     20200701105144|20f9463f-1c14-4e6...| americas/brazil/s...|rider-213|driver-213|34.158284716382845| |     20200701105144|1585ad3a-11c9-43c...|    asia/india/chennai|rider-213|driver-213|17.851135255091155| |     20200701105144|d40daa90-cf1a-4d1...|    asia/india/chennai|rider-213|driver-213|41.06290929046368| +-------------------+--------------------+----------------------+---------+----------+------------------+
　　由于测试数据分区是 区域/国家/城市，所以load(basePath ＂/*/*/*/*＂)。  修改数据
　　类似于插入新数据，使用数据生成器生成新数据对历史数据进行更新。将数据加载到DataFrame中并将DataFrame写入Hudi表中。  scala>val updates = convertToStringList(dataGen.generateUpdates(10)) scala>val df = spark.read.json(spark.sparkContext.parallelize(updates, 2)) scala>df.write.format(＂hudi＂).      |  options(getQuickstartWriteConfigs).      |  option(PRECOMBINE_FIELD_OPT_KEY, ＂ts＂).      |  option(RECORDKEY_FIELD_OPT_KEY, ＂uuid＂).      |  option(PARTITIONPATH_FIELD_OPT_KEY, ＂partitionpath＂).      |  option(TABLE_NAME, tableName).      |  mode(Append).      |  save(basePath)增量查询
　　Hudi还提供了获取自给定提交时间戳以来以更改记录流的功能。这可以通过使用Hudi的增量查询并提供开始流进行更改的开始时间来实现。  scala>spark.      |  read.      |  format(＂hudi＂).      |  load(basePath + ＂/*/*/*/*＂).      |  createOrReplaceTempView(＂hudi_trips_snapshot＂) scala>val commits = spark.sql(＂select distinct(_hoodie_commit_time) ascommitTime from  hudi_trips_snapshotorder by commitTime＂).map(k => k.getString(0)).take(50) scala>val beginTime = commits(commits.length - 2) beginTime:String = 20200701105144 scala>val tripsIncrementalDF = spark.read.format(＂hudi＂).      |  option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL).      |  option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).      |  load(basePath) scala>tripsIncrementalDF.createOrReplaceTempView(＂hudi_trips_incremental＂) scala>spark.sql(＂select `_hoodie_commit_time`, fare, begin_lon, begin_lat, tsfrom  hudi_trips_incremental where fare> 20.0＂).show() +-------------------+------------------+--------------------+-------------------+---+ |_hoodie_commit_time|              fare|           begin_lon|          begin_lat| ts| +-------------------+------------------+--------------------+-------------------+---+ |     20200701110546|49.527694252432056|  0.5142184937933181| 0.7340133901254792|0.0| |     20200701110546|  90.9053809533154| 0.19949323322922063|0.18294079059016366|0.0| |     20200701110546|  98.3428192817987|  0.3349917833248327| 0.4777395067707303|0.0| |     20200701110546| 90.25710109008239|  0.4006983139989222|0.08528650347654165|0.0| |     20200701110546| 63.72504913279929|   0.888493603696927| 0.6570857443423376|0.0| |     20200701110546|29.47661370147079|0.010872312870502165| 0.1593867607188556|0.0| +-------------------+------------------+--------------------+-------------------+---+
　　这将提供在beginTime提交后的数据，并且fare>20的数据。  时间点查询
　　根据特定时间查询，可以将endTime指向特定时间，beginTime指向000（表示最早提交时间）。  scala>val beginTime = ＂000＂ beginTime:String = 000   scala>val endTime = commits(commits.length - 2) endTime:String = 20200701105144 scala>val tripsPointInTimeDF = spark.read.format(＂hudi＂).      |  option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL).      |  option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).      |  option(END_INSTANTTIME_OPT_KEY, endTime).      |  load(basePath) scala>tripsPointInTimeDF.createOrReplaceTempView(＂hudi_trips_point_in_time＂) scala>spark.sql(＂select `_hoodie_commit_time`, fare, begin_lon, begin_lat, tsfrom hudi_trips_point_in_time where fare > 20.0＂).show() +-------------------+------------------+-------------------+-------------------+---+ |_hoodie_commit_time|              fare|          begin_lon|          begin_lat| ts| +-------------------+------------------+-------------------+-------------------+---+ |     20200701105144| 64.27696295884016|0.4923479652912024| 0.5731835407930634|0.0| |     20200701105144| 33.92216483948643|0.9694586417848392| 0.1856488085068272|0.0| |     20200701105144| 27.79478688582596|0.6273212202489661|0.11488393157088261|0.0| |     20200701105144|93.56018115236618|0.14285051259466197|0.21624150367601136|0.0| |     20200701105144|  43.4923811219014| 0.8779402295427752|0.6100070562136587|0.0| |     20200701105144|66.62084366450246|0.03844104444445928| 0.0750588760043035|0.0| |    20200701105144|34.158284716382845|0.46157858450465483| 0.4726905879569653|0.0| |     20200701105144| 41.06290929046368|0.8192868687714224| 0.651058505660742|0.0| +-------------------+------------------+-------------------+-------------------+---+删除数据  scala>spark.sql(＂select uuid, partitionPath fromhudi_trips_snapshot＂).count() res12:Long = 10 scala>val ds = spark.sql(＂select uuid, partitionPath fromhudi_trips_snapshot＂).limit(2) scala>val deletes = dataGen.generateDeletes(ds.collectAsList()) scala>val df = spark.read.json(spark.sparkContext.parallelize(deletes, 2)); scala>df.write.format(＂hudi＂).      |  options(getQuickstartWriteConfigs).      |  option(OPERATION_OPT_KEY,＂delete＂).      |  option(PRECOMBINE_FIELD_OPT_KEY, ＂ts＂).      |  option(RECORDKEY_FIELD_OPT_KEY, ＂uuid＂).      |  option(PARTITIONPATH_FIELD_OPT_KEY, ＂partitionpath＂).      |  option(TABLE_NAME, tableName).      |  mode(Append).      |  save(basePath) scala>val roAfterDeleteViewDF = spark.      |  read.      |  format(＂hudi＂).      |  load(basePath + ＂/*/*/*/*＂) scala>roAfterDeleteViewDF.registerTempTable(＂hudi_trips_snapshot＂) scala>spark.sql(＂select uuid, partitionPath fromhudi_trips_snapshot＂).count() res15:Long = 8
　　只有append模式，才支持删除功能。
　　大数据技术生态体系
　　大数据的切片机制有哪些
　　大数据之Kafka集群部署
　　大数据JUC面试题
　　大数据学习之部署Hadoop