图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › 数据湖 › 连载型› 带大家一起学数据湖：Apache+Hudi入门指南
分享

带大家一起学数据湖：Apache+Hudi入门指南

levycui 2021-1-6 20:30:32 发表于 连载型 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 2616
About云VIP帮助找工作和提升
问题导读：
1、什么是Apache Hudi？
2、Hudi如何编译？
3、Hive和Presto如何进行集成？
4、如何编写Hudi代码？


上一篇：带大家一起学数据湖：数据湖入门必备

1. 什么是Apache Hudi
  •     一个spark 库
  •     大数据更新解决方案，大数据中没有传统意义的更新，只有append和重写(Hudi就是采用重写方式)

使用Hudi的优点
  •     使用Bloomfilter机制+二次查找，可快速确定记录是更新还是新增
  •     更新范围小，是文件级别，不是表级别
  •     文件大小与hdfs的Blocksize保持一致
  •     数据文件使用parquet格式，充分利用列存的优势（dremal论文实现）
  •     提供了可扩展的大数据更新框架
  •     并发度由spark控制

hudi详细介绍见hudi官网 http://hudi.apache.org/cn/docs/0.5.0-quick-start-guide.html


2. Hudi编译
  1. 

  2. git clone https://github.com/apache/incubator-hudi.git && cd incubator-hudi

  3. mvn clean package -DskipTests -DskipITs
复制代码

注意: 本文编译hudi使用的linux环境，window环境一定要加上-DskipITs，不然会编译docker文件启动服务运行linux命令导致报错，如果是linux环境且需要用docker进行测试可以考虑去掉其参数。

3. 前置环境安装准备

所有版本选择均是查看当前master分支pom 中所依赖的 spark,hive ,hadoop,presto版本。(hudi-0.5.2-SNAPSHOT)
2021-01-06_203529.jpg

**注意：**小版本不一样不影响使用，如果运行spark任务报错不兼容排下依赖包就好。

4. Hive和Presto集成
4.1 hive

hive 查询hudi 数据主要是在hive中建立外部表数据路径指向hdfs 路径，同时hudi 重写了inputformat 和outpurtformat。因为hudi 在读的数据的时候会读元数据来决定我要加载那些parquet文件，而在写的时候会写入新的元数据信息到hdfs路径下。所以hive 要集成hudi 查询要把编译的jar 包放到HIVE-HOME/lib 下面。否则查询时找不到inputformat和outputformat的类。

hive 外表数据结构如下：
  1. CREATE EXTERNAL TABLE `test_partition`(

  2.   `_hoodie_commit_time` string, 

  3.   `_hoodie_commit_seqno` string, 

  4.   `_hoodie_record_key` string, 

  5.   `_hoodie_file_name` string, 

  6.   `id` string, 

  7.   `oid` string, 

  8.   `name` string, 

  9.   `dt` string, 

  10.   `isdeleted` string, 

  11.   `lastupdatedttm` string, 

  12.   `rowkey` string)

  13. PARTITIONED BY ( 

  14.   `_hoodie_partition_path` string)

  15. ROW FORMAT SERDE 

  16.   'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 

  17. STORED AS INPUTFORMAT 

  18.   'org.apache.hudi.hadoop.HoodieParquetInputFormat' 

  19. OUTPUTFORMAT 

  20.   'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'

  21. LOCATION

  22.   'hdfs://hj:9000/tmp/hudi'

  23. TBLPROPERTIES (

  24.   'transient_lastDdlTime'='1582111004')
复制代码

hive集成hudi方法：将hudi jar复制到hive lib下
  1. cp ./packaging/hudi-hadoop-mr-bundle/target/hudi-hadoop-mr-bundle-0.5.2-SNAPSHOT.jar  $HIVE_HOME/lib

复制代码

4.2 Presto

presto 集成hudi 是基于hive catalog 同样是访问hive 外表进行查询，如果要集成需要把hudi 包copy 到presto hive-hadoop2插件下面。
presto集成hudi方法: 将hudi jar复制到 presto hive-hadoop2下
  1. 

  2. cp  ./packaging/hudi-hadoop-mr-bundle/target/hudi-hadoop-mr-bundle-0.5.2-SNAPSHOT.jar  $PRESTO_HOME/plugin/hive-hadoop2/ 

复制代码


5. Hudi代码实战
5.1 Copy_on_Write 模式操作(默认模式)
5.1.1 insert操作（初始化插入数据）
  1. // 不带分区写入

  2.   @Test

  3.   def insert(): Unit = {

  4.     val spark = SparkSession.builder.appName("hudi insert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  5.     val insertData = spark.read.parquet("/tmp/1563959377698.parquet")

  6.     insertData.write.format("org.apache.hudi")

  7.       // 设置主键列名

  8.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  9.       // 设置数据更新时间的列名

  10.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  11.       // 并行度参数设置

  12.       .option("hoodie.insert.shuffle.parallelism", "2")

  13.       .option("hoodie.upsert.shuffle.parallelism", "2")

  14.       // table name 设置

  15.       .option(HoodieWriteConfig.TABLE_NAME, "test")

  16.       .mode(SaveMode.Overwrite)

  17.       // 写入路径设置

  18.       .save("/tmp/hudi")

  19.   }

  20. 

  21. // 带分区写入

  22.   @Test

  23.   def insertPartition(): Unit = {

  24.     val spark = SparkSession.builder.appName("hudi insert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  25.     // 读取文本文件转换为df

  26.     val insertData = Util.readFromTxtByLineToDf(spark, "/home/huangjing/soft/git/experiment/hudi-test/src/main/resources/test_insert_data.txt")

  27.     insertData.write.format("org.apache.hudi")

  28.       // 设置主键列名

  29.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  30.       // 设置数据更新时间的列名

  31.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  32.       // 设置分区列

  33.       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "dt")

  34.       // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM 四种索引 为了保证分区变更后能找到必须设置全局GLOBAL_BLOOM

  35.       .option(HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH, "true")

  36.       // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM 四种索引

  37.       .option(HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndex.IndexType.GLOBAL_BLOOM.name())

  38.       // 并行度参数设置

  39.       .option("hoodie.insert.shuffle.parallelism", "2")

  40.       .option("hoodie.upsert.shuffle.parallelism", "2")

  41.       .option(HoodieWriteConfig.TABLE_NAME, "test_partition")

  42.       .mode(SaveMode.Overwrite)

  43.       .save("/tmp/hudi")

  44.   }
复制代码




5.1.2 upsert操作（数据存在时修改，不存在时新增）
  1. // 不带分区upsert

  2.   @Test

  3.   def upsert(): Unit = {

  4. 

  5.     val spark = SparkSession.builder.appName("hudi upsert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  6.     val insertData = spark.read.parquet("/tmp/1563959377699.parquet")

  7. 

  8.     insertData.write.format("org.apache.hudi")

  9.       // 设置主键列名

  10.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  11.       // 设置数据更新时间的列名

  12.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  13.       // 表名称设置

  14.       .option(HoodieWriteConfig.TABLE_NAME, "test")

  15.       // 并行度参数设置

  16.       .option("hoodie.insert.shuffle.parallelism", "2")

  17.       .option("hoodie.upsert.shuffle.parallelism", "2")

  18.       .mode(SaveMode.Append)

  19.       // 写入路径设置

  20.       .save("/tmp/hudi");

  21.   }

  22. 

  23. // 带分区upsert

  24.   @Test

  25.   def upsertPartition(): Unit = {

  26. 

  27.     val spark = SparkSession.builder.appName("upsert partition").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  28.     val upsertData = Util.readFromTxtByLineToDf(spark, "/home/huangjing/soft/git/experiment/hudi-test/src/main/resources/test_update_data.txt")

  29. 

  30.     upsertData.write.format("org.apache.hudi").option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  31.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  32.        // 分区列设置

  33.       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "dt")

  34.       .option(HoodieWriteConfig.TABLE_NAME, "test_partition")

  35.       .option(HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndex.IndexType.GLOBAL_BLOOM.name())

  36.       .option("hoodie.insert.shuffle.parallelism", "2")

  37.       .option("hoodie.upsert.shuffle.parallelism", "2")

  38.       .mode(SaveMode.Append)

  39.       .save("/tmp/hudi");

  40.   }

复制代码



5.1.3 delete操作（删除数据）
  1.   @Test

  2.   def delete(): Unit = {

  3.     val spark = SparkSession.builder.appName("delta insert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  4.     val deleteData = spark.read.parquet("/tmp/1563959377698.parquet")

  5.     deleteData.write.format("com.uber.hoodie")

  6.       // 设置主键列名

  7.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  8.       // 设置数据更新时间的列名

  9.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  10.       // 表名称设置

  11.       .option(HoodieWriteConfig.TABLE_NAME, "test")

  12.       // 硬删除配置

  13.       .option(DataSourceWriteOptions.PAYLOAD_CLASS_OPT_KEY, "org.apache.hudi.EmptyHoodieRecordPayload")

  14.   }
复制代码




删除操作分为软删除和硬删除配置在这里查看：http://hudi.apache.org/cn/docs/0 ... 4%E6%95%B0%E6%8D%AE

5.1.4 query操作（查询数据）

  1.   @Test

  2.   def query(): Unit = {

  3.     val basePath = "/tmp/hudi"

  4.     val spark = SparkSession.builder.appName("query insert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  5.     val tripsSnapshotDF = spark.

  6.       read.

  7.       format("org.apache.hudi").

  8.       load(basePath + "/*/*")

  9. 

  10.     tripsSnapshotDF.show()

  11.   }
复制代码




5.1.5 同步至Hive
  1.   @Test

  2.   def hiveSync(): Unit = {

  3.     val spark = SparkSession.builder.appName("delta hiveSync").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  4.     val upsertData = Util.readFromTxtByLineToDf(spark, "/home/huangjing/soft/git/experiment/hudi-test/src/main/resources/hive_sync.txt")

  5. 

  6.     upsertData.write.format("org.apache.hudi")

  7.       // 设置主键列名

  8.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  9.       // 设置数据更新时间的列名

  10.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  11.       // 分区列设置

  12.       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "dt")

  13.       // 设置要同步的hive库名

  14.       .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY, "hj_repl")

  15.       // 设置要同步的hive表名

  16.       .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, "test_partition")

  17.       // 设置数据集注册并同步到hive

  18.       .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true")

  19.       // 设置当分区变更时，当前数据的分区目录是否变更

  20.       .option(HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH, "true")

  21.       // 设置要同步的分区列名

  22.       .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY, "dt")

  23.       // 设置jdbc 连接同步

  24.       .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY, "jdbc:hive2://localhost:10000")

  25.       // hudi表名称设置

  26.       .option(HoodieWriteConfig.TABLE_NAME, "test_partition")

  27.       // 用于将分区字段值提取到Hive分区列中的类,这里我选择使用当前分区的值同步

  28.       .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY, "org.apache.hudi.hive.MultiPartKeysValueExtractor")

  29.       // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM 四种索引 为了保证分区变更后能找到必须设置全局GLOBAL_BLOOM

  30.       .option(HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndex.IndexType.GLOBAL_BLOOM.name())

  31.       // 并行度参数设置

  32.       .option("hoodie.insert.shuffle.parallelism", "2")

  33.       .option("hoodie.upsert.shuffle.parallelism", "2")

  34.       .mode(SaveMode.Append)

  35.       .save("/tmp/hudi");

  36.   }

  37. 

  38.   @Test

  39.   def hiveSyncMergeOnReadByUtil(): Unit = {

  40.     val args: Array[String] = Array("--jdbc-url", "jdbc:hive2://hj:10000", "--partition-value-extractor", "org.apache.hudi.hive.MultiPartKeysValueExtractor", "--user", "hive", "--pass", "hive", "--partitioned-by", "dt", "--base-path", "/tmp/hudi_merge_on_read", "--database", "hj_repl", "--table", "test_partition_merge_on_read")

  41.     HiveSyncTool.main(args)

  42.   }
复制代码




这里可以选择使用spark 或者hudi-hive包中的hiveSynTool进行同步，hiveSynTool类其实就是run_sync_tool.sh运行时调用的。hudi 和hive同步时保证hive目标表不存在,同步其实就是建立外表的过程。

5.1.6 Hive查询读优化视图和增量视图
  1.   @Test

  2.   def hiveViewRead(): Unit = {

  3.     // 目标表

  4.     val sourceTable = "test_partition"

  5.     // 增量视图开始时间点

  6.     val fromCommitTime = "20200220094506"

  7.     // 获取当前增量视图后几个提交批次

  8.     val maxCommits = "2"

  9. 

  10.     Class.forName("org.apache.hive.jdbc.HiveDriver")

  11.     val prop = new Properties()

  12.     prop.put("user", "hive")

  13.     prop.put("password", "hive")

  14.     val conn = DriverManager.getConnection("jdbc:hive2://localhost:10000/hj_repl", prop)

  15.     val stmt = conn.createStatement

  16.     // 这里设置增量视图参数

  17.     stmt.execute("set hive.input.format=org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat")

  18.     // Allow queries without partition predicate

  19.     stmt.execute("set hive.strict.checks.large.query=false")

  20.     // Dont gather stats for the table created

  21.     stmt.execute("set hive.stats.autogather=false")

  22.     // Set the hoodie modie

  23.     stmt.execute("set hoodie." + sourceTable + ".consume.mode=INCREMENTAL")

  24.     // Set the from commit time

  25.     stmt.execute("set hoodie." + sourceTable + ".consume.start.timestamp=" + fromCommitTime)

  26.     // Set number of commits to pull

  27.     stmt.execute("set hoodie." + sourceTable + ".consume.max.commits=" + maxCommits)

  28. 

  29.     val rs = stmt.executeQuery("select * from " + sourceTable)

  30.     val metaData = rs.getMetaData

  31.     val count = metaData.getColumnCount

  32. 

  33. 

  34.     while (rs.next()) {

  35.       for (i <- 1 to count) {

  36.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  37.       }

  38.       println("-----------------------------------------------------------")

  39.     }

  40. 

  41.     rs.close()

  42.     stmt.close()

  43.     conn.close()

  44. 

  45.   }
复制代码



读优化视图即去掉增量视图参数即可。

5.1.7 Presto查询读优化视图（暂不支持增量视图）

  1.   @Test

  2.   def prestoViewRead(): Unit = {

  3.     // 目标表

  4.     val sourceTable = "test_partition"

  5.     Class.forName("com.facebook.presto.jdbc.PrestoDriver")

  6.     val conn = DriverManager.getConnection("jdbc:presto://hj:7670/hive/hj_repl", "hive", null)

  7.     val stmt = conn.createStatement

  8.     val rs = stmt.executeQuery("select * from  " + sourceTable)

  9.     val metaData = rs.getMetaData

  10.     val count = metaData.getColumnCount

  11. 

  12.     while (rs.next()) {

  13.       for (i <- 1 to count) {

  14.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  15.       }

  16.       println("-----------------------------------------------------------")

  17.     }

  18. 

  19.     rs.close()

  20.     stmt.close()

  21.     conn.close()

  22.   }
复制代码




5.2 Merge_On_Read 模式操作

5.2.1 insert操作（插入数据）

  1.   @Test

  2.   def insertPartitionMergeOnRead(): Unit = {

  3.     val spark = SparkSession.builder.appName("hudi insert").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  4.     // 读取文本文件转换为df

  5.     val insertData = Util.readFromTxtByLineToDf(spark, "/home/huangjing/soft/git/experiment/hudi-test/src/main/resources/test_insert_data.txt")

  6.     insertData.write.format("org.apache.hudi")

  7.       .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)

  8.       // 设置主键列名

  9.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  10.       // 设置数据更新时间的列名

  11.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  12.       // 设置分区列

  13.       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "dt")

  14.       // 设置当分区变更时，当前数据的分区目录是否变更

  15.       .option(HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH, "true")

  16.       // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM 四种索引 为了保证分区变更后能找到必须设置全局GLOBAL_BLOOM

  17.       .option(HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndex.IndexType.GLOBAL_BLOOM.name())

  18.       // 并行度参数设置

  19.       .option("hoodie.insert.shuffle.parallelism", "2")

  20.       .option("hoodie.upsert.shuffle.parallelism", "2")

  21.       .option(HoodieWriteConfig.TABLE_NAME, "test_partition_merge_on_read")

  22.       .mode(SaveMode.Overwrite)

  23.       .save("/tmp/hudi_merge_on_read")

  24.   }
复制代码




merge on read 主要是要是加入option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)参数，其他修改删除操作和copy on write 类似,这里不一一列举。

5.2.2 同步至Hive

  1.   @Test

  2.   def hiveSyncMergeOnRead(): Unit = {

  3.     val spark = SparkSession.builder.appName("delta hiveSync").config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").master("local[3]").getOrCreate()

  4.     val upsertData = Util.readFromTxtByLineToDf(spark, "/home/huangjing/soft/git/experiment/hudi-test/src/main/resources/hive_sync.txt")

  5. 

  6.     upsertData.write.format("org.apache.hudi")

  7.       // 配置读时合并

  8.       .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)

  9.       // 设置主键列名

  10.       .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "rowkey")

  11.       // 设置数据更新时间的列名

  12.       .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "lastupdatedttm")

  13.       // 分区列设置

  14.       .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "dt")

  15.       // 设置要同步的hive库名

  16.       .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY, "hj_repl")

  17.       // 设置要同步的hive表名

  18.       .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, "test_partition_merge_on_read")

  19.       // 设置数据集注册并同步到hive

  20.       .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true")

  21.       // 设置当分区变更时，当前数据的分区目录是否变更

  22.       .option(HoodieIndexConfig.BLOOM_INDEX_UPDATE_PARTITION_PATH, "true")

  23.       // 设置要同步的分区列名

  24.       .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY, "dt")

  25.       // 设置jdbc 连接同步

  26.       .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY, "jdbc:hive2://localhost:10000")

  27.       // hudi表名称设置

  28.       .option(HoodieWriteConfig.TABLE_NAME, "test_partition_merge_on_read")

  29.       // 用于将分区字段值提取到Hive分区列中的类,这里我选择使用当前分区的值同步

  30.       .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY, "org.apache.hudi.hive.MultiPartKeysValueExtractor")

  31.       // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM 四种索引 为了保证分区变更后能找到必须设置全局GLOBAL_BLOOM

  32.       .option(HoodieIndexConfig.INDEX_TYPE_PROP, HoodieIndex.IndexType.GLOBAL_BLOOM.name())

  33.       // 并行度参数设置

  34.       .option("hoodie.insert.shuffle.parallelism", "2")

  35.       .option("hoodie.upsert.shuffle.parallelism", "2")

  36.       .mode(SaveMode.Append)

  37.       .save("/tmp/hudi_merge_on_read");

  38.   }
复制代码




与copy on write 操作一样，不同的是merge on read 会生成两个表后缀为_ro和_rt的外表。_ro为读优化视图，_rt为实时视图。

5.2.3 Hive查询读优化视图（后缀_ro）和实时视图查询 (后缀_rt)

  1. /**

  2.    * merge on read 实时视图查询

  3.    */

  4.   @Test

  5.   def mergeOnReadRealtimeViewByHive(): Unit = {

  6.     // 目标表

  7.     val sourceTable = "test_partition_merge_on_read_rt"

  8. 

  9.     Class.forName("org.apache.hive.jdbc.HiveDriver")

  10.     val prop = new Properties()

  11.     prop.put("user", "hive")

  12.     prop.put("password", "hive")

  13.     val conn = DriverManager.getConnection("jdbc:hive2://localhost:10000/hj_repl", prop)

  14.     val stmt = conn.createStatement

  15. 

  16.     val rs = stmt.executeQuery("select * from " + sourceTable)

  17.     val metaData = rs.getMetaData

  18.     val count = metaData.getColumnCount

  19. 

  20. 

  21.     while (rs.next()) {

  22.       for (i <- 1 to count) {

  23.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  24.       }

  25.       println("-----------------------------------------------------------")

  26.     }

  27. 

  28.     rs.close()

  29.     stmt.close()

  30.     conn.close()

  31.   }

  32. 

  33. 

  34.   /**

  35.    * merge on read 读优化视图查询

  36.    */

  37.   @Test

  38.   def mergeOnReadReadoptimizedViewByHive(): Unit = {

  39.     // 目标表

  40.     val sourceTable = "test_partition_merge_on_read_ro"

  41. 

  42.     Class.forName("org.apache.hive.jdbc.HiveDriver")

  43.     val prop = new Properties()

  44.     prop.put("user", "hive")

  45.     prop.put("password", "hive")

  46.     val conn = DriverManager.getConnection("jdbc:hive2://localhost:10000/hj_repl", prop)

  47.     val stmt = conn.createStatement

  48. 

  49.     val rs = stmt.executeQuery("select * from " + sourceTable)

  50.     val metaData = rs.getMetaData

  51.     val count = metaData.getColumnCount

  52. 

  53. 

  54.     while (rs.next()) {

  55.       for (i <- 1 to count) {

  56.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  57.       }

  58.       println("-----------------------------------------------------------")

  59.     }

  60. 

  61.     rs.close()

  62.     stmt.close()

  63.     conn.close()

  64.   }

复制代码



5.2.4 Presto查询读优化视图（后缀_ro）和实时视图查询 (后缀_rt)

  1. /**

  2.    * presto merge on read 实时视图查询

  3.    */

  4.   @Test

  5.   def mergeOnReadRealtimeViewByPresto(): Unit = {

  6.     // 目标表

  7.     val sourceTable = "test_partition_merge_on_read_rt"

  8.     Class.forName("com.facebook.presto.jdbc.PrestoDriver")

  9.     val conn = DriverManager.getConnection("jdbc:presto://hj:7670/hive/hj_repl", "hive", null)

  10.     val stmt = conn.createStatement

  11.     val rs = stmt.executeQuery("select * from  " + sourceTable)

  12.     val metaData = rs.getMetaData

  13.     val count = metaData.getColumnCount

  14. 

  15.     while (rs.next()) {

  16.       for (i <- 1 to count) {

  17.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  18.       }

  19.       println("-----------------------------------------------------------")

  20.     }

  21. 

  22.     rs.close()

  23.     stmt.close()

  24.     conn.close()

  25.   }

  26. 

  27. 

  28.   /**

  29.    * presto merge on read 读优化视图查询

  30.    */

  31.   @Test

  32.   def mergeOnReadReadoptimizedViewByPresto(): Unit = {

  33.     // 目标表

  34.     val sourceTable = "test_partition_merge_on_read_ro"

  35.     Class.forName("com.facebook.presto.jdbc.PrestoDriver")

  36.     val conn = DriverManager.getConnection("jdbc:presto://hj:7670/hive/hj_repl", "hive", null)

  37.     val stmt = conn.createStatement

  38.     val rs = stmt.executeQuery("select * from  " + sourceTable)

  39.     val metaData = rs.getMetaData

  40.     val count = metaData.getColumnCount

  41. 

  42.     while (rs.next()) {

  43.       for (i <- 1 to count) {

  44.         println(metaData.getColumnName(i) + ":" + rs.getObject(i).toString)

  45.       }

  46.       println("-----------------------------------------------------------")

  47.     }

  48. 

  49.     rs.close()

  50.     stmt.close()

  51.     conn.close()

  52.   }
复制代码




6. 问题整理
a. merg on read 问题
merge on read 要配置option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)才会生效，配置为option(HoodieTableConfig.HOODIE_TABLE_TYPE_PROP_NAME, HoodieTableType.MERGE_ON_READ.name())将不会生效。


b. spark pom 依赖问题
不要引入spark-hive 的依赖里面包含了hive 1.2.1的相关jar包，而hudi 要求的版本是2.x版本。如果一定要使用请排除相关依赖。

c. hive视图同步问题

代码与hive视图同步时resources要加入hive-site.xml 配置文件,不然同步hive metastore 会报错。
git 测试代码地址:https://github.com/hj2016/hudi-test

作者：别过来胖到我了
来源：https://blog.csdn.net/h335146502/article/details/104485494/

最新经典文章，欢迎关注公众号


回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
levycui

超级版主

关注

  • 653

    主题

  • 1166

    帖子

  • 97

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000