Spark integrates Iceberg

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

Case

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

Plan

Create Spark SQL query job and download dependencies

• iceberg-spark-runtime-3.4_2.12-1.6.1.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.sql.autoBroadcastJoinThreshold": "-1",
  "spark.sql.catalog.hive_catalog": "org.apache.iceberg.spark.SparkCatalog",
  "spark.sql.catalog.hive_catalog.clients": "10",
  "spark.sql.catalog.hive_catalog.type": "hive",
  "spark.sql.catalog.hive_catalog.uri": "thrift://172.19.189.247:9083",
  "spark.sql.catalog.hive_catalog.warehouse": "hdfs://172.19.189.247:9000/user/hive/warehouse",
  "spark.sql.catalog.spark_catalog": "org.apache.iceberg.spark.SparkSessionCatalog",
  "spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
  "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": "STRICT"
}

SQL addition, deletion, modification and query data

-- 建iceberg表
CREATE TABLE IF NOT EXISTS  hive_catalog.ispong_db.iceberg_table (
     username string,
     age bigint
) USING iceberg;

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

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

-- 查
select * from hive_catalog.ispong_db.iceberg_table;