Spark integrates Hudi

This article provides an implementation plan for Spark integrated Hudi scenarios, including configuration examples, execution steps, and parameter descriptions.

Case

Use spark to add, delete, modify and query Hudi data

Note

The hive client needs to download the hudi-hadoop-mr-bundle-1.0.2.Jar package to access hudi data. There is no need to restart the hive service, just restart the hive session.

• hudi-hadoop-mr-bundle-1.0.2.jar download

mv hudi-hadoop-mr-bundle-1.0.2.jar /opt/hive/lib/
mv hudi-hadoop-mr-bundle-1.0.2.jar /opt/hadoop/share/hadoop/yarn/lib/

solution

Create Spark SQL query job and download dependencies

• hudi-spark3.4-bundle_2.12-1.0.2.jar download

ConfigureSparkConfig

{
  "spark.cores.max": "1",
  "spark.driver.cores": "1",
  "spark.driver.extraJavaOptions": "-Dfile.encoding=utf-8",
  "spark.driver.memory": "1g",
  "spark.executor.cores": "1",
  "spark.executor.extraJavaOptions": "-Dfile.encoding=utf-8",
  "spark.executor.instances": "1",
  "spark.executor.memory": "2g",
  "spark.hadoop.hive.metastore.uris": "thrift://172.19.189.246:9083",
  "spark.kryo.registrator": "org.apache.spark.HoodieSparkKryoRegistrar",
  "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
  "spark.sql.autoBroadcastJoinThreshold": "-1",
  "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog",
  "spark.sql.catalog.spark_catalog.type": "hudi",
  "spark.sql.catalogImplementation": "hive",
  "spark.sql.extensions": "org.apache.spark.sql.hudi.HoodieSparkSessionExtension",
  "spark.sql.hive.convertMetastoreOrc": "false",
  "spark.sql.legacy.timeParserPolicy": "LEGACY",
  "spark.sql.parquet.datetimeRebaseModeInRead": "LEGACY",
  "spark.sql.parquet.datetimeRebaseModeInWrite": "LEGACY",
  "spark.sql.parquet.enableVectorizedReader": "false",
  "spark.sql.parquet.int96RebaseModeInRead": "LEGACY",
  "spark.sql.parquet.int96RebaseModeInWrite": "LEGACY",
  "spark.sql.parquet.writeLegacyFormat": "true",
  "spark.sql.storeAssignmentPolicy": "LEGACY"
}

SQL addition, deletion, modification and query data

-- 选择catalog
use spark_catalog;

-- 建hudi表
CREATE TABLE IF NOT EXISTS ispong_db.hudi_table (
    username STRING,
    age BIGINT
) USING HUDI;

-- 增
insert into ispong_db.hudi_table values ('zhangsan',13);
insert into ispong_db.hudi_table values ('lisi',14);

-- 删
delete from ispong_db.hudi_table where username = 'lisi';
    
-- 改
update ispong_db.hudi_table set age = 133 where username ='zhangsan'; 

-- 查
select username ,age  from ispong_db.hudi_table;