• 大数据下一代变革之必研究数据湖技术Hudi原理实战双管齐下-下

    集成Spark开发

    Spark编程读写示例

    通过IDE如Idea编程实质上和前面的spark-shell和spark-sql相似，其他都是Spark编程的知识，下面以scala语言为示例，idea新建scala的maven项目

    image-20221124110101979

    pom文件添加如下依赖

    1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    2.   <modelVersion>4.0.0</modelVersion>
    3.   <groupId>cn.itxs</groupId>
    4.   <artifactId>hoodie-spark-demo</artifactId>
    5.   <version>1.0</version>
    6.  
    7.   <properties>
    8.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    9.     <scala.version>2.12.10</scala.version>
    10.     <scala.binary.version>2.12</scala.binary.version>
    11.     <spark.version>3.3.0</spark.version>
    12.     <hoodie.version>0.12.1</hoodie.version>
    13.     <hadoop.version>3.3.4</hadoop.version>
    14.   </properties>
    15.  
    16.   <dependencies>
    17.     <dependency>
    18.       <groupId>org.scala-lang</groupId>
    19.       <artifactId>scala-library</artifactId>
    20.       <version>${scala.version}</version>
    21.     </dependency>
    22.  
    23.     <dependency>
    24.       <groupId>org.apache.spark</groupId>
    25.       <artifactId>spark-core_${scala.binary.version}</artifactId>
    26.       <version>${spark.version}</version>
    27.       <scope>provided</scope>
    28.     </dependency>
    29.  
    30.     <dependency>
    31.       <groupId>org.apache.spark</groupId>
    32.       <artifactId>spark-sql_${scala.binary.version}</artifactId>
    33.       <version>${spark.version}</version>
    34.       <scope>provided</scope>
    35.     </dependency>
    36.  
    37.     <dependency>
    38.       <groupId>org.apache.spark</groupId>
    39.       <artifactId>spark-hive_${scala.binary.version}</artifactId>
    40.       <version>${spark.version}</version>
    41.       <scope>provided</scope>
    42.     </dependency>
    43.  
    44.     <dependency>
    45.       <groupId>org.apache.hadoop</groupId>
    46.       <artifactId>hadoop-client</artifactId>
    47.       <version>${hadoop.version}</version>
    48.       <scope>provided</scope>
    49.     </dependency>
    50.  
    51.     <dependency>
    52.       <groupId>org.apache.hudi</groupId>
    53.       <artifactId>hudi-spark3.3-bundle_${scala.binary.version}</artifactId>
    54.       <version>${hoodie.version}</version>
    55.       <scope>provided</scope>
    56.     </dependency>
    57.   </dependencies>
    58.  
    59.   <build>
    60.     <plugins>
    61.       <plugin>
    62.         <groupId>org.apache.maven.plugins</groupId>
    63.         <artifactId>maven-compiler-plugin</artifactId>
    64.         <version>3.10.1</version>
    65.         <configuration>
    66.           <source>1.8</source>
    67.           <target>1.8</target>
    68.           <encoding>${project.build.sourceEncoding}</encoding>
    69.         </configuration>
    70.       </plugin>
    71.       <plugin>
    72.         <groupId>org.scala-tools</groupId>
    73.         <artifactId>maven-scala-plugin</artifactId>
    74.         <version>2.15.2</version>
    75.         <executions>
    76.           <execution>
    77.             <goals>
    78.               <goal>compile</goal>
    79.               <goal>testCompile</goal>
    80.             </goals>
    81.           </execution>
    82.         </executions>
    83.       </plugin>
    84.       <plugin>
    85.         <groupId>org.apache.maven.plugins</groupId>
    86.         <artifactId>maven-shade-plugin</artifactId>
    87.         <version>3.2.4</version>
    88.         <executions>
    89.           <execution>
    90.             <phase>package</phase>
    91.             <goals>
    92.               <goal>shade</goal>
    93.             </goals>
    94.             <configuration>
    95.               <filters>
    96.                 <filter>
    97.                   <artifact>*:*</artifact>
    98.                   <excludes>
    99.                     <exclude>META-INF/*.SF</exclude>
    100.                     <exclude>META-INF/*.DSA</exclude>
    101.                     <exclude>META-INF/*.RSA</exclude>
    102.                   </excludes>
    103.                 </filter>
    104.               </filters>
    105.             </configuration>
    106.           </execution>
    107.         </executions>
    108.       </plugin>
    109.     </plugins>
    110.   </build>
    111. </project>

    创建常量对象

    1. object Constant {
    2.     val HUDI_STORAGE_PATH = "hdfs://192.168.5.53:9000/tmp/"
    3. }

    插入hudi数据

    1. package cn.itxs
    2.  
    3. import org.apache.spark.sql.SparkSession
    4. import org.apache.spark.SparkConf
    5. import org.apache.hudi.QuickstartUtils._
    6. import scala.collection.JavaConversions._
    7. import org.apache.spark.sql.SaveMode._
    8. import org.apache.hudi.DataSourceWriteOptions._
    9. import org.apache.hudi.config.HoodieWriteConfig._
    10.  
    11. object InsertDemo {
    12.   def main(args: Array[String]): Unit = {
    13.     val sparkConf = new SparkConf()
    14.       .setAppName(this.getClass.getSimpleName)
    15.       .setMaster("local[*]")
    16.       .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    17.  
    18.     val sparkSession = SparkSession.builder()
    19.       .config(sparkConf)
    20.       .enableHiveSupport()
    21.       .getOrCreate()
    22.  
    23.     val tableName = "hudi_trips_cow_idea"
    24.     val basePath =  Constant.HUDI_STORAGE_PATH+tableName
    25.     val dataGen = new DataGenerator
    26.     val inserts = convertToStringList(dataGen.generateInserts(10))
    27.  
    28.     val df = sparkSession.read.json(sparkSession.sparkContext.parallelize(inserts,2))
    29.     df.write.format("hudi").
    30.       options(getQuickstartWriteConfigs).
    31.       option(PRECOMBINE_FIELD.key(), "ts").
    32.       option(RECORDKEY_FIELD.key(), "uuid").
    33.       option(PARTITIONPATH_FIELD.key(), "partitionpath").
    34.       option(TBL_NAME.key(), tableName).
    35.       mode(Overwrite).
    36.       save(basePath)
    37.  
    38.     sparkSession.close()
    39.   }
    40. }

    由于依赖中scope是配置为provided，因此运行配置中勾选下面这项

    image-20221124111557461

    运行InsertDemo程序写入hudi数据

    image-20221124111827746

    运行ReadDemo程序读取hudi数据

    image-20221124112658848

    通过mvn clean package打包后上传运行

    1. spark-submit \
    2. --class cn.itxs.ReadDemo \
    3. /home/commons/spark-3.3.0-bin-hadoop3/appjars/hoodie-spark-demo-1.0.jar

    DeltaStreamer

    HoodieDeltaStreamer实用程序(hudi-utilities-bundle的一部分)提供了从不同源(如DFS或Kafka)中获取的方法，具有以下功能。

    • 从Kafka的新事件，从Sqoop的增量导入或输出HiveIncrementalPuller或DFS文件夹下的文件。
    • 支持json, avro或自定义记录类型的传入数据。
    • 管理检查点、回滚和恢复。
    • 利用来自DFS或Confluent模式注册中心的Avro模式。
    • 支持插入转换。
    1. # 拷贝hudi-utilities-bundle_2.12-0.12.1.jar到spark的jars目录
    2. cp /home/commons/hudi-release-0.12.1/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-0.12.1.jar jars/
    3. # 查看帮助文档，参数非常多，可以在有需要使用的时候查阅
    4. spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer /home/commons/spark-3.3.0-bin-hadoop3/jars/hudi-utilities-bundle_2.12-0.12.1.jar --help

    image-20221124170418737

    该工具采用层次结构组成的属性文件，并具有提取数据、密钥生成和提供模式的可插入接口。在hudi-下提供了从kafka和dfs中摄取的示例配置

    image-20221124152601371

    接下里以File Based Schema Provider和JsonKafkaSoiurce为示例演示如何使用

    1. # 创建topic
    2. bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 --create --partitions 1 --replication-factor 1 --topic data_test

    然后编写demo程序持续向这个kafka的topic发送消息

    image-20221124152926618

    1. # 创建一个配置文件目录
    2. mkdir /home/commons/hudi-properties
    3. # 拷贝示例配置文件
    4. cp hudi-utilities/src/test/resources/delta-streamer-config/kafka-source.properties /home/commons/hudi-properties/
    5. cp hudi-utilities/src/test/resources/delta-streamer-config/base.properties /home/commons/hudi-properties/

    定义avro所需的schema文件包括source和target,创建source文件 vim source-json-schema.avsc

    1. {
    2.   "type" : "record",
    3.   "name" : "Profiles",
    4.   "fields" : [
    5.       {
    6.         "name" : "id",
    7.         "type" : "long"
    8.       }, {
    9.         "name" : "name",
    10.         "type" : "string"
    11.       }, {
    12.          "name" : "age",
    13.          "type" : "int"
    14.       },  {
    15.         "name" : "partitions",
    16.         "type" : "int"
    17.       }
    18.   ]
    19. }

    拷贝一份为target文件

    cp source-json-schema.avsc target-json-schema.avsc

    修改kafka-source.properties的配置如下

    1. include=hdfs://hadoop2:9000/hudi-properties/base.properties
    2. # Key fields, for kafka example
    3. hoodie.datasource.write.recordkey.field=id
    4. hoodie.datasource.write.partitionpath.field=partitions
    5. # schema provider configs
    6. #hoodie.deltastreamer.schemaprovider.registry.url=http://localhost:8081/subjects/impressions-value/versions/latest
    7. hoodie.deltastreamer.schemaprovider.source.schema.file=hdfs://hadoop2:9000/hudi-properties/source-json-schema.avsc
    8. hoodie.deltastreamer.schemaprovider.target.schema.file=hdfs://hadoop2:9000/hudi-properties/target-json-schema.avsc
    9. # Kafka Source
    10. #hoodie.deltastreamer.source.kafka.topic=uber_trips
    11. hoodie.deltastreamer.source.kafka.topic=data_test
    12. #Kafka props
    13. bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
    14. auto.offset.reset=earliest
    15. #schema.registry.url=http://localhost:8081
    16. group.id=mygroup

    将本地hudi-properties文件夹上传到HDFS

    1. cd ..
    2. hdfs dfs -put hudi-properties/ /

    image-20221124160153231

    1. # 运行导入命令
    2. spark-submit \
    3. --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer \
    4. /home/commons/spark-3.3.0-bin-hadoop3/jars/hudi-utilities-bundle_2.12-0.12.1.jar \
    5.   --props hdfs://hadoop2:9000/hudi-properties/kafka-source.properties \
    6.   --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider \
    7.   --source-class org.apache.hudi.utilities.sources.JsonKafkaSource \
    8.   --source-ordering-field id \
    9.   --target-base-path hdfs://hadoop2:9000/tmp/hudi/user_test \
    10.   --target-table user_test \
    11.   --op BULK_INSERT \
    12.   --table-type MERGE_ON_READ

    image-20221124171559468

    查看hdfs目录已经有表目录和分区目录

    image-20221124171723635

    image-20221124171826926

    通过spark-sql查询从kafka摄取的数据

    1. use hudi_spark;
    2. create table user_test using hudi
    3. location 'hdfs://hadoop2:9000/tmp/hudi/user_test';
    4. select * from user_test limit 10;

    image-20221124172628568

    环境准备

    1. # 解压进入flink目录，这里我就用之前flink的环境，详细可以查看之前关于flink的文章
    2. cd /home/commons/flink-1.15.1
    3. # 拷贝编译好的jar到flink的lib目录
    4. cp /home/commons/hudi-release-0.12.1/packaging/hudi-flink-bundle/target/hudi-flink1.15-bundle-0.12.1.jar lib/

    image-20221124173958802

    1. # 拷贝guava包，解决依赖冲突
    2. cp /home/commons/hadoop/share/hadoop/common/lib/guava-27.0-jre.jar lib/
    3. # 配置hadoop环境变量和启动hadoop
    4. export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

    sql-clent使用

    启动

    修改配置文件 vi conf/flink-conf.yaml

    1. classloader.check-leaked-classloader: false
    2. taskmanager.numberOfTaskSlots: 4
    3. state.backend: rocksdb
    4. state.checkpoints.dir: hdfs://hadoop2:9000/checkpoints/flink
    5. state.backend.incremental: true
    6. execution.checkpointing.interval: 5min
    • local 模式

    修改workers文件，也可以多配制几个（伪分布式或完全分布式），官方提供示例是4个

    1. localhost
    2. localhost
    3. localhost
    1. # 在本机上启动三个TaskManagerRunner和一个Standalone伪分布式集群
    2. ./bin/start-cluster.sh 
    3. # 查看进程确认
    4. jps -l

    image-20221125092325266

    1. # 启动内嵌的flink sql客户端
    2. ./bin/sql-client.sh embedded
    3. show databases;
    4. show tables;

    image-20221125092721153

    • yarn-session 模式

      • 解决依赖冲突问题
      1. # 拷贝jar到flink的lib目录
      2. cp /home/commons/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.3.4.jar lib/
      • 启动yarn-session
      1. # 先停止上面启动Standalone伪分布式集群
      2. ./bin/stop-cluster.sh
      3. # 启动yarn-session分布式集群
      4. ./bin/yarn-session.sh --detached

      image-20221125183041918

      查看yarn上已经有一个Flink session集群job， ID为application_1669357770610_0015

      image-20221125183108137

      查看Flink的Web UI可用TaskSlots为0，可确认已切换为yarn管理资源非分配

      image-20221125180205029

      • 启动sql-client
      1. # 由于使用内嵌模式管理元数据，元数据是保存在内存中，关闭sql-client后则元数据也会消失，生产环境建议使用如Hive元数据管理方式，后面再做配置
      2. ./bin/sql-client.sh embedded -s yarn-session
      3. show databases;
      4. show tables;

    插入数据

    1. CREATE TABLE t1(
    2.   uuid VARCHAR(20),
    3.   name VARCHAR(10),
    4.   age INT,
    5.   ts TIMESTAMP(3),
    6.   `partition` VARCHAR(20),
    7.   PRIMARY KEY(uuid) NOT ENFORCED
    8. )
    9. PARTITIONED BY (`partition`)
    10. WITH (
    11.   'connector' = 'hudi',
    12.   'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t1',
    13.   'table.type' = 'MERGE_ON_READ' -- 创建一个MERGE_ON_READ表，默认情况下是COPY_ON_WRITE表
    14. );
    15. -- 插入数据
    16. INSERT INTO t1 VALUES
    17.   ('id1','Danny',23,TIMESTAMP '2022-11-25 00:00:01','par1'),
    18.   ('id2','Stephen',33,TIMESTAMP '2022-11-25 00:00:02','par1'),
    19.   ('id3','Julian',53,TIMESTAMP '2022-11-25 00:00:03','par2'),
    20.   ('id4','Fabian',31,TIMESTAMP '2022-11-25 00:00:04','par2'),
    21.   ('id5','Sophia',18,TIMESTAMP '2022-11-25 00:00:05','par3'),
    22.   ('id6','Emma',20,TIMESTAMP '2022-11-25 00:00:06','par3'),
    23.   ('id7','Bob',44,TIMESTAMP '2022-11-25 00:00:07','par4'),
    24.   ('id8','Han',56,TIMESTAMP '2022-11-25 00:00:08','par4');

    image-20221128092047672

    查看Flink Web UI Job的信息

    image-20221128091855135

    image-20221128092026837

    1. # 查询数据
    2. select * from t1;

    image-20221128092459685

    1. # 更新数据
    2. INSERT INTO t1 VALUES
    3.   ('id1','Danny',28,TIMESTAMP '2022-11-25 00:00:01','par1');
    4. # 查询数据
    5. select * from t1;

    image-20221128133630036

    流式读取

    1. -- 设置结果模式为tableau，在CLI中直接显示结果;另外还有table和changelog；changelog模式可以获取+I,-U之类动作数据；
    2. set 'sql-client.execution.result-mode' = 'tableau';
    3. CREATE TABLE sourceT (
    4.   uuid varchar(20),
    5.   name varchar(10),
    6.   age int,
    7.   ts timestamp(3),
    8.   `partition` varchar(20),
    9.   PRIMARY KEY(uuid) NOT ENFORCED
    10. ) WITH (
    11.   'connector' = 'datagen',
    12.   'rows-per-second' = '1'
    13. );
    14.  
    15. CREATE TABLE t2 (
    16.   uuid varchar(20),
    17.   name varchar(10),
    18.   age int,
    19.   ts timestamp(3),
    20.   `partition` varchar(20),
    21.   PRIMARY KEY(uuid) NOT ENFORCED
    22. )
    23. WITH (
    24. 'connector' = 'hudi', 
    25. 'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t2', 
    26. 'table.type' = 'MERGE_ON_READ',
    27. 'read.streaming.enabled' = 'true',
    28. 'read.streaming.check-interval' = '4'
    29. );
    30.  
    31. insert into t2 select * from sourceT;
    32. select * from t2;

    image-20221128140741157

    image-20221128143313273

    Bucket索引

    在0.11.0增加了一种高效、轻量级的索引类型bucket index，其为字节贡献回馈给hudi社区。

    • Bucket Index是一种Hash分配方式，根据指定的索引字段，计算hash值，然后结合Bucket个数，均匀分配到具体的文件中。Bucket Index支持大数据量场景下的更新，Bucket Index也可以对数据进行分桶存储，但是对于桶数的计算是需要根据当前数据量的大小进行评估的，如果后续需要re-hash的话成本也会比较高。在这里我们预计通过建立Extensible Hash Index来提高哈希索引的可扩展能力。
    • 要使用此索引，请将索引类型设置为BUCKET并设置hoodie.storage.layout.partitioner.class为org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner。对于 Flink，设置index.type=BUCKET.
    • 该方式相比于BloomIndex在元素定位性能高很多，缺点是Bucket个数无法动态扩展。另外Bucket不适合于COW表，否则会导致写放大更严重。
    • 实时入湖写入的性能要求高的场景建议采用Bucket索引。

    Hudi Catalog

    前面基于内容管理hudi元数据的方式每次重启sql客户端就丢掉了，Hudi Catalog则是可以持久化元数据；Hudi Catalog支持多种模式，包括dfs和hms,hudi还可以直接集群hive使用，后续再一步步演示，现在先简单看下dfs模式的Hudi Catalog，先添加启动sql文件，vim conf/sql-client-init.sql

    1. create catalog hudi_catalog 
    2. with(
    3. 'type' = 'hudi',
    4. 'mode' = 'dfs',
    5. 'catalog.path'='/tmp/hudi_catalog'
    6. );
    7. use catalog hudi_catalog;

    创建目录并启动，建表测试

    1. hdfs dfs -mkdir /tmp/hudi_catalog
    2. ./bin/sql-client.sh embedded -i conf/sql-client-init.sql -s yarn-session

    image-20221128183632702

    查看hdfs的数据如下，退出客户端后重新登录客户端还可以查到上面的hudi_catalog及其库和表的数据。

    image-20221128183822461

  • 相关阅读:
    AuNPs功能化聚苯乙烯丙烯酸微球/导电聚苯胺纳米线/聚苯胺纳米纤维修饰聚苯乙烯微球制备方法
    58 乘积最大子数组
    k8s从入门到精通
    2022-11-14 mysql-loose scan
    Kepler.gl笔记：地图交互
    释放静电行为监测识别系统
    在同一局域网中使用电脑访问手机内容
    FLowUs：一款优秀强大、好看的文件协作软件
    国产+开源：可视化流程引擎助力企业建立流程管理体系
    集成树模型的应用，kaggle的信用卡忠诚度预测案例（一）
  • 原文地址：https://blog.csdn.net/jh035512/article/details/128128375