百度360必应搜狗淘宝本站头条
头像打包下载acmecadconverter_8.52绿色版mysql 昨天的日期word文档批量处理大师破解版document.appendchild
当前位置：网站首页 > 技术教程 > 正文

数据湖(五):Hudi 与 Hive 集成

csdh11 2024-11-30 14:14 4 浏览

#头条创作挑战赛#

Hudi与Hive集成

一、配置HiveServer2

Hudi与Hive集成原理是通过代码方式将数据写入到HDFS目录中，那么同时映射Hive表，让Hive表映射的数据对应到此路径上，这时Hudi需要通过JDBC方式连接Hive进行元数据操作，这时需要配置HiveServer2。

1、在Hive服务端配置hive-site.xml

#在Hive 服务端 $HIVE_HOME/conf/hive-site.xml中配置： <!-- 配置hiveserver2 --> <property> <name>hive.server2.thrift.port</name> <value>10000</value> </property> <property> <name>hive.server2.thrift.bind.host</name> <value>192.168.179.4</value> </property> <!-- 配置hiveserver2使用的zookeeper --> <property> <name>hive.zookeeper.quorum</name> <value> node3:2181,node4:2181,node5:2181</value> </property>

注意：“hive.zookeeper.quorum”搭建hiveserver2HA使用配置项，可以不配置，如果不配置启动hiveServer2时一直连接本地zookeeper,导致大量错误日志（/tmp/root/hive.log），从而导致通过beeline连接当前node1节点的hiveserver2时不稳定，会有连接不上错误信息。

2、在每台Hadoop 节点配置core-site.xml,记得发送到所有节点

<!-- 配置代理访问用户，如果不配置下列信息 hive的jdbc连接会报错 --> <property> <name>hadoop.proxyuser.root.hosts</name> <value>*</value> </property> <property> <name>hadoop.proxyuser.root.groups</name> <value>*</value> </property>

3、重启HDFS ，Hive ，在Hive服务端启动Metastore 和 HiveServer2服务

[root@node1 conf]# hive --service metastore &
[root@node1 conf]# hive --service hiveserver2 &

4、在客户端通过beeline连接Hive

[root@node3 test]# beeline
beeline> !connect jdbc:hive2://node1:10000 root
Enter password for jdbc:hive2://node1:10000: **** #可以输入任意密码,没有验证
0: jdbc:hive2://node1:10000> show tables;

二、代码层面集成Hudi与Hive

我们可以通过SparkSQL将数据保存到Hudi中同时也映射到Hive表中。映射有两种模式，如果Hudi表是COPY_ON_WRITE类型，那么映射成的Hive表对应是指定的Hive表名，此表中存储着Hudi所有数据。

如果Hudi表类型是MERGE_ON_READ模式，那么映射的Hive表将会有2张，一张后缀为rt ，另一张表后缀为ro。后缀rt对应的Hive表中存储的是Base文件Parquet格式数据+log Avro格式数据，也就是全量数据。后缀为ro Hive表中存储的是存储的是Base文件对应的数据。

1）在pom.xml中加入一下依赖

<!-- 连接Hive 驱动包--> <dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-jdbc</artifactId> <version>1.2.1</version> </dependency>

2）将对应依赖包放入Hive节点对应的lib目录下

将hudi-hadoop-mr-bundle-0.8.0.jar、parquet-column-1.10.1.jar、parquet-common-1.10.1.jar、parquet-format-2.4.0.jar、parquet-hadoop-1.10.1.jar包存入Hive lib目录下。由于Hudi表数据映射到Hive表后，Hive表底层存储格式为“HoodieParquetInputFormat”或者“HoodieParquetRealtimeInputFormat”，解析Parquet数据格式时使用到以上各个包。可以从Maven中下载以上包后，将这些包上传到所有Hive节点的lib目录下，包括服务端和客户端。

3）启动Hive MetaStore与Hive Server2服务

[root@node1 conf]# hive --service metastore &
[root@node1 conf]# hive --service hiveserver2 &

4）将hive-site.xml放入项目resources目录中

后期Hudi映射Hive表后，会自动检查Hive元数据，这时需要找到配置文件连接Hive。

1、COW模式-SparkSQL代码写入Hudi同时映射Hive表

1）COW模式代码如下

//1.创建对象
val session: SparkSession = SparkSession.builder().master("local").appName("insertDataToHudi")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

//2.创建DataFrame
val insertDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\jsondata.json")

import org.apache.spark.sql.functions._
//3.向Hudi中插入数据 - COW模式
insertDF
  .write.format("hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
  //设置主键列名称
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  //当数据主键相同时，对比的字段，保存该字段大的数据
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  //指定分区列
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "loc")
  //并行度设置
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  //表名设置
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos")
  //关于Hive设置
  //指定HiveServer2 连接url
  .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://node1:10000")
  //指定Hive 对应的库名
  .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
  //指定Hive映射的表名称
  .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"infos1")
  //Hive表映射对的分区字段
  .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"loc")
  //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
  .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
  //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
  .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos")

2）查询Hive中数据

hive> show tables;
infos1
hive> set hive.cli.print.header=true;

hive> select * from infos1;
hive> select `_hoodie_commit_time`,id,name,age,loc,data_dt from infos1;

3）更新表中数据,再次查询Hive中的数据

//4.更新数据，查询Hive数据
//读取修改数据
val updateDataDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\updatedata.json")

//向Hudi 更新数据
updateDataDF.write.format("org.apache.hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"loc")
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos")
  //关于Hive设置
  //指定HiveServer2 连接url
  .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://node1:10000")
  //指定Hive 对应的库名
  .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
  //指定Hive映射的表名称
  .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"infos1")
  //Hive表映射对的分区字段
  .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"loc")
  //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
  .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
  //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
  .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos")

每次查询都是查询最后一次数据结果

2、MOR模式-SparkSQL代码写入Hudi同时映射Hive表

1）MOR代码如下

//1.创建对象
val session: SparkSession = SparkSession.builder().master("local").appName("insertDataToHudi")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

//2.创建DataFrame
val insertDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\jsondata.json")

import org.apache.spark.sql.functions._
//3.向Hudi中插入数据 - COW模式
insertDF
  .write.format("hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
  //设置主键列名称
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  //当数据主键相同时，对比的字段，保存该字段大的数据
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  //指定分区列
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "loc")
  //并行度设置
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  //表名设置,不能重复，重复会报错
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos2")
  //关于Hive设置
  //指定HiveServer2 连接url
  .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://node1:10000")
  //指定Hive 对应的库名
  .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
  //指定Hive映射的表名称
  .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"infos2")
  //Hive表映射对的分区字段
  .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"loc")
  //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
  .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
  //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
  .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos2")

2）查询Hive表中的数据

hive> show tables;

注意：infos2_ro 中存储的只是Base文件中数据（parquet列式存储结果）

infos2_rt 中存储的是Base文件(Parquet列式存储结果)+*log*(Avro行式存储结果)

hive> select * from infos2_ro;

目前只有Base文件数据，查询的就是全量数据

hive> select * from infos2_rt;

目前只有Base文件数据，查询的就是全量数据

hive> select `_hoodie_commit_time`,id,name,age,loc,data_dt from infos2_ro;
hive> select `_hoodie_commit_time`,id,name,age,loc,data_dt from infos2_rt;

3）更新表中数据,再次查询Hive中的数据

//4.更新数据，查询Hive数据
//读取修改数据
val updateDataDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\updatedata.json")

//向Hudi 更新数据，注意，必须指定Hive对应配置
updateDataDF.write.format("org.apache.hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"loc")
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos2")
  //关于Hive设置
  //指定HiveServer2 连接url
  .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://node1:10000")
  //指定Hive 对应的库名
  .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
  //指定Hive映射的表名称
  .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"infos2")
  //Hive表映射对的分区字段
  .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"loc")
  //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
  .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
  //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
  .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos2")

查询Hive中对应两张表的数据结果

查询Base文件中的数据

hive> select `_hoodie_commit_time`,id,name,age,loc,data_dt from infos2_ro;

查询Base文件+log文件中数据，可以看到查询到的是修改后的结果数据

hive> select `_hoodie_commit_time`,id,name,age,loc,data_dt from infos2_rt;

三、手动层面集成Hudi与Hive

如果已经存在Hudi数据，我们也可以手动创建对应的Hive表来映射对应的Hudi数据，使用Hive SQL来操作Hudi。例如使用如下代码在HDFS中存储Hudi数据，这里采用MOR模式写入数据，方便后期测试：

1）向Hudi表中写入数据

//1.创建对象
val session: SparkSession = SparkSession.builder().master("local").appName("insertDataToHudi")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .getOrCreate()

//2.创建DataFrame
val insertDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\jsondata.json")

import org.apache.spark.sql.functions._
//3.向Hudi中插入数据 - COW模式
insertDF
  .write.format("hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
  //设置主键列名称
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  //当数据主键相同时，对比的字段，保存该字段大的数据
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  //指定分区列
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "loc")
  //并行度设置
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  //表名设置
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos3")
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos3")

2）在Hive中创建对应的表数据

在Hive中创建表person3_ro，映射Base数据，相当于前面的ro表：

// 创建外部表,这种方式只会查询出来parquet数据文件中的内容，但是刚刚更新或者删除的数据不能查出来
CREATE EXTERNAL TABLE `person3_ro`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`id` bigint,`name` string,
`age` bigint,`data_dt` string)
PARTITIONED BY (`loc` string) 
ROW FORMAT SERDE 
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
'org.apache.hudi.hadoop.HoodieParquetInputFormat' 
OUTPUTFORMAT 
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' 
LOCATION 'hdfs://mycluster/hudi_data/person_infos3';

建好以上对应的表之后，由于有分区，还看不到数据，所以这里需要我们手动映射分区数据：

alter table person3_ro add if not exists partition(loc="beijing") location 'hdfs://mycluster/hudi_data/person_infos3/beijing';
alter table person3_ro add if not exists partition(loc='chongqing') location 'hdfs://mycluster/hudi_data/person_infos3/chongqing';
alter table person3_ro add if not exists partition(loc='hainai') location 'hdfs://mycluster/hudi_data/person_infos3/hainai';
alter table person3_ro add if not exists partition(loc='hunan') location 'hdfs://mycluster/hudi_data/person_infos3/hunan';
alter table person3_ro add if not exists partition(loc='shandong') location 'hdfs://mycluster/hudi_data/person_infos3/shandong';
alter table person3_ro add if not exists partition(loc='shanghai') location 'hdfs://mycluster/hudi_data/person_infos3/shanghai';
alter table person3_ro add if not exists partition(loc='shenzhen') location 'hdfs://mycluster/hudi_data/person_infos3/shenzhen';
alter table person3_ro add if not exists partition(loc='tianjin') location 'hdfs://mycluster/hudi_data/person_infos3/tianjin';

查看表数据

在Hive中创建表person3_rt，映射Base+log数据，相当于rt表，并映射分区：

// 这种方式会将基于Parquet的基础列式文件、和基于行的Avro日志文件合并在一起呈现给用户。
CREATE EXTERNAL TABLE `person3_rt`(
`_hoodie_commit_time` string,
`_hoodie_commit_seqno` string,
`_hoodie_record_key` string,
`_hoodie_partition_path` string,
`_hoodie_file_name` string,
`id` bigint,`name` string,
`age` bigint,`data_dt` string)
PARTITIONED BY (`loc` string) 
ROW FORMAT SERDE 
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat' 
OUTPUTFORMAT 
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' 
LOCATION 'hdfs://mycluster/hudi_data/person_infos3';

加载对应分区数据数据

alter table person3_rt add if not exists partition(loc="beijing") location 'hdfs://mycluster/hudi_data/person_infos3/beijing';
alter table person3_rt add if not exists partition(loc='chongqing') location 'hdfs://mycluster/hudi_data/person_infos3/chongqing';
alter table person3_rt add if not exists partition(loc='hainai') location 'hdfs://mycluster/hudi_data/person_infos3/hainai';
alter table person3_rt add if not exists partition(loc='hunan') location 'hdfs://mycluster/hudi_data/person_infos3/hunan';
alter table person3_rt add if not exists partition(loc='shandong') location 'hdfs://mycluster/hudi_data/person_infos3/shandong';
alter table person3_rt add if not exists partition(loc='shanghai') location 'hdfs://mycluster/hudi_data/person_infos3/shanghai';
alter table person3_rt add if not exists partition(loc='shenzhen') location 'hdfs://mycluster/hudi_data/person_infos3/shenzhen';
alter table person3_rt add if not exists partition(loc='tianjin') location 'hdfs://mycluster/hudi_data/person_infos3/tianjin';

查看结果数据

3）使用代码修改Hudi表中的数据：

修改数据如下：

{"id":1,"name":"ls1","age":40,"loc":"beijing","data_dt":"20210709"}
{"id":2,"name":"ls2","age":50,"loc":"shanghai","data_dt":"20210710"}
{"id":3,"name":"ls3","age":60,"loc":"ttt","data_dt":"20210711"}
//4.更新数据，查询Hive数据
//读取修改数据
val updateDataDF: DataFrame = session.read.json("file:///D:\\2018IDEA_space\\SparkOperateHudi\\data\\updatedata.json")

//向Hudi 更新数据
updateDataDF.write.format("org.apache.hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"loc")
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos3")
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos3")

4）继续查询对应的两张Hive表数据

由于分区 “ttt”是新加入的分区，需要手动添加下分区才能在对应的Hive表中正常查询

alter table person3_ro add if not exists partition(loc="ttt") location 'hdfs://mycluster/hudi_data/person_infos3/ttt';
alter table person3_rt add if not exists partition(loc="ttt") location 'hdfs://mycluster/hudi_data/person_infos3/ttt';

查询表person3_ro

hive> select * from person3_ro;

查询表person3_rt

hive> select * from person3_rt;

此外，我们也可以不需要每次都自己手动添加分区，而是创建好对应的Hive表后，在代码中向Hudi中写数据时，指定对应的Hive参数即可，这样写入的数据自动会映射到Hive中。

我们可以删除Hive对应的表数据重新创建以及第一次加载分区，再后续写入Hudi表数据时，代码如下，就不需要每次都手动加载Hive分区数据。

//5.更新数据，指定Hive配置项
//读取修改数据
val updateDataDF: DataFrame = session.read.json("file:///D:\\2022IDEA_space\\SparkOperateHudi\\data\\updatedata.json")
//向Hudi 更新数据
updateDataDF.write.format("org.apache.hudi")
  //设置写出模式，默认就是COW
  .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY,DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
  .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "id")
  .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "data_dt")
  .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"loc")
  .option("hoodie.insert.shuffle.parallelism", "2")
  .option("hoodie.upsert.shuffle.parallelism", "2")
  .option(HoodieWriteConfig.TABLE_NAME, "person_infos3")
  //关于Hive设置
  //指定HiveServer2 连接url
  .option(DataSourceWriteOptions.HIVE_URL_OPT_KEY,"jdbc:hive2://node1:10000")
  //指定Hive 对应的库名
  .option(DataSourceWriteOptions.HIVE_DATABASE_OPT_KEY,"default")
  //指定Hive映射的表名称
  .option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY,"person3")
  //Hive表映射对的分区字段
  .option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,"loc")
  //当设置为true时，注册/同步表到Apache Hive metastore,默认是false，这里就是自动创建表
  .option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY,"true")
  //如果分区格式不是yyyy/mm/dd ，需要指定解析类将分区列解析到Hive中
  .option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,classOf[MultiPartKeysValueExtractor].getName)
  .mode(SaveMode.Append)
  .save("/hudi_data/person_infos3")

查询Hive表 peson3_ro数据

查询Hive表 person3_rt数据

四、SparkSQL操作映射的Hive表

将Hudi数据映射到Hive后，我们可以使用SparkSQL来进行操作Hive表进行处理。操作代码如下：

val session: SparkSession = SparkSession.builder().master("local").appName("test")
  .config("hive.metastore.uris", "thrift://node1:9083")
  .enableHiveSupport()
  .getOrCreate()

session.sql("use default")
session.sql(
  """
    | select id,name,age,loc,data_dt from person3_rt
  """.stripMargin).show()
session.sql(
  """
    | select sum(age) as totalage from person3_rt
  """.stripMargin).show()

我们可以看到如果针对Hudi表中的数据进行修改，那么SpakSQL读取到的就是最新修改后的结果数据。

相关推荐

Micheal Nielsen&#39;s神经网络学习之二

依然是跟着MichaelNielsen的神经网络学习，基于前一篇的学习，已经大概明白了神经网络的基本结构和BP算法，也能通过神经网络训练数字识别功能，之后我试验了一下使用神经网络训练之前的文本分类，...

CocoaPods + XCTest进行单元测试 c单元测试工具

在使用XCTest进行单元测试时，我们经常会遇到一些CocoaPods中的开源框架的调用，比如“Realm”或“Alamofire”在测试的时候，如果配置不当，会导致“frameworknotfo...

Java基础知识回顾第四篇 java基础讲解

1、&和&&的区别作为逻辑运算符：&(不管左边是什么，右边都参与运算)，&&(如果左边为false，右边则不参与运算，短路)另外&可作为位运算符...

项目中的流程及类似业务的设计模式总结

说到业务流程，可能是我做过的项目中涉及业务最多的一个方面了。除了在流程设计之外，在一些考核系统、产业审批、还有很多地方，都用到相似的设计思路，在此一并总结一下。再说到模式，并不是因为流行才用这个词，而...

联想三款显示器首批获得 Eyesafe Certified 2.0 认证

IT之家7月31日消息，据外媒报道，三款全新联想显示器是全球首批满足EyesafeCertified2.0的设备。据报道，联想获得EyesafeCertified2.0认证的显...

maven的生命周期,插件介绍(二) 一个典型的maven构建生命周期

1.maven生命周期一个完整的项目构建过程通常包括清理、编译、测试、打包、集成测试、验证、部署等步骤，Maven从中抽取了一套完善的、易扩展的生命周期。Maven的生命周期是抽象的，其中的具体任务都...

多线程(3)-基于Object的线程等待与唤醒

概述在使用synchronized进行线程同步中介绍了依赖对象锁定线程，本篇文章介绍如何依赖对象协调线程。同synchronized悲观锁一样，线程本身不能等待与唤醒，也是需要对象才能完成等待与唤醒的...

jquery mobile + 百度地图 + phonegap 写的一个&quot;校园助手&quot;的app

1jquerymobile+百度地图+phonegap写的一个"校园助手"的app，使用的是基于Flat-UI的jQueryMobile，请参考：https://github.com/...

Apache 服务启动不了 apache系统服务启动不了

{我是新手，从未遇到此问题，请各位大大勿喷}事由：今天早上上班突然发现公司网站出现问题。经过排查，发现是Apache出现问题。首先检查配置文件没有出问题后，启动服务发现Apache服务能启动，但是没法...

健康债和技术债都不能欠 公众号: 我是攻城师(woshigcs)

在Solr4.4之后，Solr提供了SolrCloud分布式集群的模式，它带来的主要好处是：（1）大数据量下更高的性能（2）更好扩展性（3）更高的可靠性（4）更简单易用什么时候应该使用Sol...

Eye Experience怎么用?HTC告诉你 eyebeam怎么用

IT之家（www.ithome.com）：EyeExperience怎么用？HTC告诉你HTC上周除了发布HTCDesireEYE自拍机和HTCRE管状运动相机之外，还发布了一系列新的智能手机...

Android系统应用隐藏和应用禁止卸载

1、应用隐藏与禁用Android设置中的应用管理器提供了一个功能，就是【应用停用】功能，这是针对某些系统应用的。当应用停用之后，应用的图标会被隐藏，但apk还是存在，不会删除，核心接口就是Packag...

计算机软件技术分享--赠人玫瑰,手遗余香

一、Netty介绍Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。也就是说，Netty...

Gecco爬虫框架的线程和队列模型 爬虫通用框架

简述爬虫在抓取一个页面后一般有两个任务，一个是解析页面内容，一个是将需要继续抓取的url放入队列继续抓取。因此，当爬取的网页很多的情况下，待抓取url的管理也是爬虫框架需要解决的问题。本文主要说的是g...

一点感悟(一) 初识 初读感知的意思

时间过得很快，在IT业已从业了两年多。人这一辈子到底需要什么，在路边看着人来人往，大部分人脸上都是很匆忙。上海真是一个魔都，它有魅力，有底蕴，但是一个外地人在这里扎根置业，真的是举全家之力，还贷3...

一周热门
最近发表
标签列表