This guide provides a quick peek at Hudi’s capabilities using Spark. Using Spark Datasource APIs(both scala and python) and using Spark SQL, we will walk through code snippets that allows you to insert, update, delete and query a Hudi table.

Setup

Hudi works with Spark-2.4.3+ & Spark 3.x versions. You can follow instructions here for setting up Spark.

Spark 3 Support Matrix

Hudi Supported Spark 3 version
0.15.x 3.5.x (default build), 3.4.x, 3.3.x, 3.2.x, 3.1.x, 3.0.x
0.14.x 3.4.x (default build), 3.3.x, 3.2.x, 3.1.x, 3.0.x
0.13.x 3.3.x (default build), 3.2.x, 3.1.x
0.12.x 3.3.x (default build), 3.2.x, 3.1.x
0.11.x 3.2.x (default build, Spark bundle only), 3.1.x
0.10.x 3.1.x (default build), 3.0.x
0.7.0 - 0.9.0 3.0.x
0.6.0 and prior not supported

The default build Spark version indicates how we build hudi-spark3-bundle.

Change summary

In 0.15.0, we introduced the support for Spark 3.5.x. In 0.14.0, we introduced the support for Spark 3.4.x and bring back the support for Spark 3.0.x. In 0.12.0, we introduced the experimental support for Spark 3.3.0. In 0.11.0, there are changes on using Spark bundles, please refer to 0.11.0 release notes for detailed instructions.

Spark Shell/SQL

Scala

From the extracted directory run spark-shell with Hudi:

  1. # For Spark versions: 3.2 - 3.5
  2. export SPARK_VERSION=3.5 # or 3.4, 3.3, 3.2
  3. spark-shell --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 \
  4. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  5. --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
  6. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  7. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark versions: 3.0 - 3.1
  2. export SPARK_VERSION=3.1 # or 3.0
  3. spark-shell --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 \
  4. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  5. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  6. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark version: 2.4
  2. spark-shell --packages org.apache.hudi:hudi-spark2.4-bundle_2.11:0.15.0 \
  3. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  4. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  5. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'

Python

From the extracted directory run pyspark with Hudi:

  1. # For Spark versions: 3.2 - 3.5
  2. export PYSPARK_PYTHON=$(which python3)
  3. export SPARK_VERSION=3.5 # or 3.4, 3.3, 3.2
  4. pyspark --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark versions: 3.0 - 3.1
  2. export PYSPARK_PYTHON=$(which python3)
  3. export SPARK_VERSION=3.1 # or 3.0
  4. pyspark --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark version: 2.4
  2. export PYSPARK_PYTHON=$(which python3)
  3. pyspark --packages org.apache.hudi:hudi-spark2.4-bundle_2.11:0.15.0 --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'

Spark SQL

Hudi support using Spark SQL to write and read data with the HoodieSparkSessionExtension sql extension. From the extracted directory run Spark SQL with Hudi:

  1. # For Spark versions: 3.2 - 3.5
  2. export SPARK_VERSION=3.5 # or 3.4, 3.3, 3.2
  3. spark-sql --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 \
  4. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  5. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  6. --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
  7. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark versions: 3.0 - 3.1
  2. export SPARK_VERSION=3.1 # or 3.0
  3. spark-sql --packages org.apache.hudi:hudi-spark$SPARK_VERSION-bundle_2.12:0.15.0 \
  4. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  5. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  6. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'
  1. # For Spark version: 2.4
  2. spark-sql --packages org.apache.hudi:hudi-spark2.4-bundle_2.11:0.15.0 \
  3. --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
  4. --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
  5. --conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'

on Kryo serialization

Users are recommended to set this config to reduce Kryo serialization overhead

  1. --conf 'spark.kryo.registrator=org.apache.spark.HoodieKryoRegistrar'

for Spark 3.2 and higher versions

Use scala 2.12 builds with an additional config:

  1. --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

Setup project

Below, we do imports and setup the table name and corresponding base path.

Scala

  1. // spark-shell
  2. import scala.collection.JavaConversions._
  3. import org.apache.spark.sql.SaveMode._
  4. import org.apache.hudi.DataSourceReadOptions._
  5. import org.apache.hudi.DataSourceWriteOptions._
  6. import org.apache.hudi.common.table.HoodieTableConfig._
  7. import org.apache.hudi.config.HoodieWriteConfig._
  8. import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
  9. import org.apache.hudi.common.model.HoodieRecord
  10. import spark.implicits._
  12. val tableName = "trips_table"
  13. val basePath = "file:///tmp/trips_table"

Python

  1. # pyspark
  2. from pyspark.sql.functions import lit, col
  4. tableName = "trips_table"
  5. basePath = "file:///tmp/trips_table"

Spark SQL

  1. // Next section will go over create table commands

Create Table

First, let’s create a Hudi table. Here, we use a partitioned table for illustration, but Hudi also supports non-partitioned tables.

Scala

  1. // scala
  2. // First commit will auto-initialize the table, if it did not exist in the specified base path.

Python

  1. # pyspark
  2. # First commit will auto-initialize the table, if it did not exist in the specified base path.

Spark SQL

For users who have Spark-Hive integration in their environment, this guide assumes that you have the appropriate settings configured to allow Spark to create tables and register in Hive Metastore.

Here is an example of creating a Hudi table.

  1. -- create a Hudi table that is partitioned.
  2. CREATE TABLE hudi_table (
  3.     ts BIGINT,
  4.     uuid STRING,
  5.     rider STRING,
  6.     driver STRING,
  7.     fare DOUBLE,
  8.     city STRING
  9. ) USING HUDI
  10. PARTITIONED BY (city);

For more options for creating Hudi tables or if you’re running into any issues, please refer to SQL DDL reference guide.

Insert data

Scala

Generate some new records as a DataFrame and write the DataFrame into a Hudi table. Since, this is the first write, it will also auto-create the table.

  1. // spark-shell
  2. val columns = Seq("ts","uuid","rider","driver","fare","city")
  3. val data =
  4.   Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
  5.     (1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),
  6.     (1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),
  7.     (1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo"    ),
  8.     (1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"));
  10. var inserts = spark.createDataFrame(data).toDF(columns:_*)
  11. inserts.write.format("hudi").
  12.   option("hoodie.datasource.write.partitionpath.field", "city").
  13.   option("hoodie.table.name", tableName).
  14.   mode(Overwrite).
  15.   save(basePath)

Mapping to Hudi write operations

Hudi provides a wide range of write operations - both batch and incremental - to write data into Hudi tables, with different semantics and performance. When record keys are not configured (see keys below), bulk_insert will be chosen as the write operation, matching the out-of-behavior of Spark’s Parquet Datasource.

Python

Generate some new records as a DataFrame and write the DataFrame into a Hudi table. Since, this is the first write, it will also auto-create the table.

  1. # pyspark
  2. columns = ["ts","uuid","rider","driver","fare","city"]
  3. data =[(1695159649087,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
  4.        (1695091554788,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),
  5.        (1695046462179,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),
  6.        (1695516137016,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo"),
  7.        (1695115999911,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai")]
  8. inserts = spark.createDataFrame(data).toDF(*columns)
  10. hudi_options = {
  11.     'hoodie.table.name': tableName,
  12.     'hoodie.datasource.write.partitionpath.field': 'city'
  13. }
  15. inserts.write.format("hudi"). \
  16.     options(**hudi_options). \
  17.     mode("overwrite"). \
  18.     save(basePath)

Mapping to Hudi write operations Hudi provides a wide range of write operations - both batch and incremental - to write data into Hudi tables, with different semantics and performance. When record keys are not configured (see keys below), bulk_insert will be chosen as the write operation, matching the out-of-behavior of Spark’s Parquet Datasource.

Spark SQL

Users can use ‘INSERT INTO’ to insert data into a Hudi table. See Insert Into for more advanced options.

  1. INSERT INTO hudi_table
  2. VALUES
  3. (1695159649087,'334e26e9-8355-45cc-97c6-c31daf0df330','rider-A','driver-K',19.10,'san_francisco'),
  4. (1695091554788,'e96c4396-3fad-413a-a942-4cb36106d721','rider-C','driver-M',27.70 ,'san_francisco'),
  5. (1695046462179,'9909a8b1-2d15-4d3d-8ec9-efc48c536a00','rider-D','driver-L',33.90 ,'san_francisco'),
  6. (1695332066204,'1dced545-862b-4ceb-8b43-d2a568f6616b','rider-E','driver-O',93.50,'san_francisco'),
  7. (1695516137016,'e3cf430c-889d-4015-bc98-59bdce1e530c','rider-F','driver-P',34.15,'sao_paulo'    ),
  8. (1695376420876,'7a84095f-737f-40bc-b62f-6b69664712d2','rider-G','driver-Q',43.40 ,'sao_paulo'    ),
  9. (1695173887231,'3eeb61f7-c2b0-4636-99bd-5d7a5a1d2c04','rider-I','driver-S',41.06 ,'chennai'      ),
  10. (1695115999911,'c8abbe79-8d89-47ea-b4ce-4d224bae5bfa','rider-J','driver-T',17.85,'chennai');

If you want to control the Hudi write operation used for the INSERT statement, you can set the following config before issuing the INSERT statement:

  1. -- bulk_insert using INSERT_INTO 
  2. SET hoodie.spark.sql.insert.into.operation = 'bulk_insert'

Query data

Hudi tables can be queried back into a DataFrame or Spark SQL.

Scala

  1. // spark-shell
  2. val tripsDF = spark.read.format("hudi").load(basePath)
  3. tripsDF.createOrReplaceTempView("trips_table")
  5. spark.sql("SELECT uuid, fare, ts, rider, driver, city FROM  trips_table WHERE fare > 20.0").show()
  6. spark.sql("SELECT _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare FROM  trips_table").show()

Python

  1. # pyspark
  2. tripsDF = spark.read.format("hudi").load(basePath)
  3. tripsDF.createOrReplaceTempView("trips_table")
  5. spark.sql("SELECT uuid, fare, ts, rider, driver, city FROM  trips_table WHERE fare > 20.0").show()
  6. spark.sql("SELECT _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare FROM trips_table").show()

Spark SQL

  1.  SELECT ts, fare, rider, driver, city FROM  hudi_table WHERE fare > 20.0;

Update data

Hudi tables can be updated by streaming in a DataFrame or using a standard UPDATE statement.

Scala

  1. // Lets read data from target Hudi table, modify fare column for rider-D and update it. 
  2. val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-D").withColumn("fare", col("fare") * 10)
  4. updatesDf.write.format("hudi").
  5.   option("hoodie.datasource.write.operation", "upsert").
  6.   option("hoodie.datasource.write.partitionpath.field", "city").
  7.   option("hoodie.table.name", tableName).
  8.   mode(Append).
  9.   save(basePath)

Key requirements Updates with spark-datasource is feasible only when the source dataframe contains Hudi’s meta fields or a key field is configured. Notice that the save mode is now Append. In general, always use append mode unless you are trying to create the table for the first time.

Spark SQL

Hudi table can be update using a regular UPDATE statement. See Update for more advanced options.

  1. UPDATE hudi_table SET fare = 25.0 WHERE rider = 'rider-D';

Python

  1. # pyspark
  2. # Lets read data from target Hudi table, modify fare column for rider-D and update it.
  3. updatesDf = spark.read.format("hudi").load(basePath).filter("rider == 'rider-D'").withColumn("fare",col("fare")*10)
  5. updatesDf.write.format("hudi"). \
  6.   options(**hudi_options). \
  7.   mode("append"). \
  8.   save(basePath)

Key requirements Updates with spark-datasource is feasible only when the source dataframe contains Hudi’s meta fields or a key field is configured. Notice that the save mode is now Append. In general, always use append mode unless you are trying to create the table for the first time.

Querying the data again will now show updated records. Each write operation generates a new commit. Look for changes in _hoodie_commit_time, fare fields for the given _hoodie_record_key value from a previous commit.

Merging Data

Scala

  1. // spark-shell
  2. val adjustedFareDF = spark.read.format("hudi").
  3.   load(basePath).limit(2).
  4.   withColumn("fare", col("fare") * 10)
  5. adjustedFareDF.write.format("hudi").
  6. option("hoodie.datasource.write.payload.class","com.payloads.CustomMergeIntoConnector").
  7. mode(Append).
  8. save(basePath)
  9. // Notice Fare column has been updated but all other columns remain intact.
  10. spark.read.format("hudi").load(basePath).show()

The com.payloads.CustomMergeIntoConnector adds adjusted fare values to the original table and preserves all other fields. Refer here for sample implementation of com.payloads.CustomMergeIntoConnector.

Python

  1. # pyspark
  2. adjustedFareDF = spark.read.format("hudi").load(basePath). \
  3.     limit(2).withColumn("fare", col("fare") * 100)
  4. adjustedFareDF.write.format("hudi"). \
  5. option("hoodie.datasource.write.payload.class","com.payloads.CustomMergeIntoConnector"). \
  6. mode("append"). \
  7. save(basePath)
  8. # Notice Fare column has been updated but all other columns remain intact.
  9. spark.read.format("hudi").load(basePath).show()

The com.payloads.CustomMergeIntoConnector adds adjusted fare values to the original table and preserves all other fields. Refer here for sample implementation of com.payloads.CustomMergeIntoConnector.

Spark SQL

  1. -- source table using Hudi for testing merging into target Hudi table
  2. CREATE TABLE fare_adjustment (ts BIGINT, uuid STRING, rider STRING, driver STRING, fare DOUBLE, city STRING) 
  3. USING HUDI;
  4. INSERT INTO fare_adjustment VALUES 
  5. (1695091554788,'e96c4396-3fad-413a-a942-4cb36106d721','rider-C','driver-M',-2.70 ,'san_francisco'),
  6. (1695530237068,'3f3d9565-7261-40e6-9b39-b8aa784f95e2','rider-K','driver-U',64.20 ,'san_francisco'),
  7. (1695241330902,'ea4c36ff-2069-4148-9927-ef8c1a5abd24','rider-H','driver-R',66.60 ,'sao_paulo'    ),
  8. (1695115999911,'c8abbe79-8d89-47ea-b4ce-4d224bae5bfa','rider-J','driver-T',1.85,'chennai'      );
  11. MERGE INTO hudi_table AS target
  12. USING fare_adjustment AS source
  13. ON target.uuid = source.uuid
  14. WHEN MATCHED THEN UPDATE SET target.fare = target.fare + source.fare
  15. WHEN NOT MATCHED THEN INSERT *
  16. ;

Key requirements

  1. For a Hudi table with user defined primary record keys, the join condition is expected to contain the primary keys of the table. For a Hudi table with Hudi generated primary keys, the join condition can be on any arbitrary data columns.
  2. For Merge-On-Read tables, partial column updates are not yet supported, i.e. all columns need to be SET from a MERGE statement either using SET * or using SET column1 = expression1 [, column2 = expression2 ...].

Delete data

Delete operation removes the records specified from the table. For example, this code snippet deletes records for the HoodieKeys passed in. Check out the deletion section for more details.

Scala

  1. // spark-shell
  2. // Lets  delete rider: rider-D
  3. val deletesDF = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-F")
  5. deletesDF.write.format("hudi").
  6.   option("hoodie.datasource.write.operation", "delete").
  7.   option("hoodie.datasource.write.partitionpath.field", "city").
  8.   option("hoodie.table.name", tableName).
  9.   mode(Append).
  10.   save(basePath)

Querying the data again will not show the deleted record.

Key requirements Deletes with spark-datasource is supported only when the source dataframe contains Hudi’s meta fields or a key field is configured. Notice that the save mode is again Append.

Spark SQL

  1. DELETE FROM hudi_table WHERE uuid = '3f3d9565-7261-40e6-9b39-b8aa784f95e2';

Python

  1. # pyspark
  2. # Lets  delete rider: rider-D
  3. deletesDF = spark.read.format("hudi").load(basePath).filter("rider == 'rider-F'")
  5. # issue deletes
  6. hudi_hard_delete_options = {
  7.   'hoodie.table.name': tableName,
  8.   'hoodie.datasource.write.partitionpath.field': 'city',
  9.   'hoodie.datasource.write.operation': 'delete',
  10. }
  12. deletesDF.write.format("hudi"). \
  13. options(**hudi_hard_delete_options). \
  14. mode("append"). \
  15. save(basePath)

Querying the data again will not show the deleted record.

Key requirements Deletes with spark-datasource is supported only when the source dataframe contains Hudi’s meta fields or a key field is configured. Notice that the save mode is again Append.

Time Travel Query

Hudi supports time travel query to query the table as of a point-in-time in history. Three timestamp formats are supported as illustrated below.

Scala

  1. spark.read.format("hudi").
  2.   option("as.of.instant", "20210728141108100").
  3.   load(basePath)
  5. spark.read.format("hudi").
  6.   option("as.of.instant", "2021-07-28 14:11:08.200").
  7.   load(basePath)
  9. // It is equal to "as.of.instant = 2021-07-28 00:00:00"
  10. spark.read.format("hudi").
  11.   option("as.of.instant", "2021-07-28").
  12.   load(basePath)

Python

  1. # pyspark
  2. spark.read.format("hudi"). \
  3.   option("as.of.instant", "20210728141108100"). \
  4.   load(basePath)
  6. spark.read.format("hudi"). \
  7.   option("as.of.instant", "2021-07-28 14:11:08.000"). \
  8.   load(basePath)
  10. # It is equal to "as.of.instant = 2021-07-28 00:00:00"
  11. spark.read.format("hudi"). \
  12.   option("as.of.instant", "2021-07-28"). \
  13.   load(basePath)

Spark SQL

  2. -- time travel based on commit time, for eg: `20220307091628793`
  3. SELECT * FROM hudi_table TIMESTAMP AS OF '20220307091628793' WHERE id = 1;
  4. -- time travel based on different timestamp formats
  5. SELECT * FROM hudi_table TIMESTAMP AS OF '2022-03-07 09:16:28.100' WHERE id = 1;
  6. SELECT * FROM hudi_table TIMESTAMP AS OF '2022-03-08' WHERE id = 1;

Requires Spark 3.2+

Incremental query

Hudi provides the unique capability to obtain a set of records that changed between a start and end commit time, providing you with the “latest state” for each such record as of the end commit time. By default, Hudi tables are configured to support incremental queries, using record level metadata tracking.

Below, we fetch changes since a given begin time while the end time defaults to the latest commit on the table. Users can also specify an end time using END_INSTANTTIME.key() option.

Scala

  1. // spark-shell
  2. spark.read.format("hudi").load(basePath).createOrReplaceTempView("trips_table")
  4. val commits = spark.sql("SELECT DISTINCT(_hoodie_commit_time) AS commitTime FROM  trips_table ORDER BY commitTime").map(k => k.getString(0)).take(50)
  5. val beginTime = commits(commits.length - 2) // commit time we are interested in
  7. // incrementally query data
  8. val tripsIncrementalDF = spark.read.format("hudi").
  9.   option("hoodie.datasource.query.type", "incremental").
  10.   option("hoodie.datasource.read.begin.instanttime", 0).
  11.   load(basePath)
  12. tripsIncrementalDF.createOrReplaceTempView("trips_incremental")
  14. spark.sql("SELECT `_hoodie_commit_time`, fare, rider, driver, uuid, ts FROM  trips_incremental WHERE fare > 20.0").show()

Python

  1. # pyspark
  2. # reload data
  3. spark.read.format("hudi").load(basePath).createOrReplaceTempView("trips_table")
  5. commits = list(map(lambda row: row[0], spark.sql("SELECT DISTINCT(_hoodie_commit_time) AS commitTime FROM  trips_table ORDER BY commitTime").limit(50).collect()))
  6. beginTime = commits[len(commits) - 2] # commit time we are interested in
  8. # incrementally query data
  9. incremental_read_options = {
  10.   'hoodie.datasource.query.type': 'incremental',
  11.   'hoodie.datasource.read.begin.instanttime': beginTime,
  12. }
  14. tripsIncrementalDF = spark.read.format("hudi"). \
  15.   options(**incremental_read_options). \
  16.   load(basePath)
  17. tripsIncrementalDF.createOrReplaceTempView("trips_incremental")
  19. spark.sql("SELECT `_hoodie_commit_time`, fare, rider, driver, uuid, ts FROM trips_incremental WHERE fare > 20.0").show()

Spark SQL

  1. -- syntax
  2. hudi_table_changes(table or path, queryType, beginTime [, endTime]);  
  3. -- table or path: table identifier, example: db.tableName, tableName, 
  4. --                or path for of your table, example: path/to/hudiTable  
  5. --                in this case table does not need to exist in the metastore,
  6. -- queryType: incremental query mode, example: latest_state, cdc  
  7. --            (for cdc query, first enable cdc for your table by setting cdc.enabled=true),
  8. -- beginTime: instantTime to begin query from, example: earliest, 202305150000, 
  9. -- endTime: optional instantTime to end query at, example: 202305160000, 
  11. -- incrementally query data by table name
  12. -- start from earliest available commit, end at latest available commit.  
  13. SELECT * FROM hudi_table_changes('db.table', 'latest_state', 'earliest');
  15. -- start from earliest, end at 202305160000.  
  16. SELECT * FROM hudi_table_changes('table', 'latest_state', 'earliest', '202305160000');  
  18. -- start from 202305150000, end at 202305160000.
  19. SELECT * FROM hudi_table_changes('table', 'latest_state', '202305150000', '202305160000');

Change Data Capture Query

Hudi also exposes first-class support for Change Data Capture (CDC) queries. CDC queries are useful for applications that need to obtain all the changes, along with before/after images of records, given a commit time range.

Scala

  1. // spark-shell
  2. // Lets first insert data to a new table with cdc enabled.
  3. val columns = Seq("ts","uuid","rider","driver","fare","city")
  4. val data =
  5.   Seq((1695158649187L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
  6.     (1695091544288L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-B","driver-L",27.70 ,"san_paulo"),
  7.     (1695046452379L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-C","driver-M",33.90 ,"san_francisco"),
  8.     (1695332056404L,"1dced545-862b-4ceb-8b43-d2a568f6616b","rider-D","driver-N",93.50,"chennai"));
  9. var df = spark.createDataFrame(data).toDF(columns:_*)
  11. // Insert data
  12. df.write.format("hudi").
  13.   option("hoodie.datasource.write.partitionpath.field", "city").
  14.   option("hoodie.table.cdc.enabled", "true").
  15.   option("hoodie.table.name", tableName).
  16.   mode(Overwrite).
  17.   save(basePath)
  19. // Update fare for riders: rider-A and rider-B 
  20. val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-A" || $"rider" === "rider-B").withColumn("fare", col("fare") * 10)
  22. updatesDf.write.format("hudi").
  23.   option("hoodie.datasource.write.operation", "upsert").
  24.   option("hoodie.datasource.write.partitionpath.field", "city").
  25.   option("hoodie.table.cdc.enabled", "true").
  26.   option("hoodie.table.name", tableName).
  27.   mode(Append).
  28.   save(basePath)
  31. // Query CDC data
  32. spark.read.option("hoodie.datasource.read.begin.instanttime", 0).
  33.   option("hoodie.datasource.query.type", "incremental").
  34.   option("hoodie.datasource.query.incremental.format", "cdc").
  35.   format("hudi").load(basePath).show(false)

Python

  1. # pyspark
  2. # Lets first insert data to a new table with cdc enabled.
  3. columns = ["ts","uuid","rider","driver","fare","city"]
  4. data =[(1695159649087,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
  5.        (1695091554788,"e96c4396-3fad-413a-a942-4cb36106d721","rider-B","driver-L",27.70 ,"san_francisco"),
  6.        (1695046462179,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-C","driver-M",33.90 ,"san_francisco"),
  7.        (1695516137016,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-C","driver-N",34.15,"sao_paulo")]
  10. inserts = spark.createDataFrame(data).toDF(*columns)
  12. hudi_options = {
  13.     'hoodie.table.name': tableName,
  14.     'hoodie.datasource.write.partitionpath.field': 'city',
  15.     'hoodie.table.cdc.enabled': 'true'
  16. }
  17. # Insert data
  18. inserts.write.format("hudi"). \
  19.     options(**hudi_options). \
  20.     mode("overwrite"). \
  21.     save(basePath)
  24. #  Update fare for riders: rider-A and rider-B 
  25. updatesDf = spark.read.format("hudi").load(basePath).filter("rider == 'rider-A' or rider == 'rider-B'").withColumn("fare",col("fare")*10)
  27. updatesDf.write.format("hudi"). \
  28.   mode("append"). \
  29.   save(basePath)
  31. # Query CDC data
  32. cdc_read_options = {
  33.     'hoodie.datasource.query.incremental.format': 'cdc',
  34.     'hoodie.datasource.query.type': 'incremental',
  35.     'hoodie.datasource.read.begin.instanttime': 0
  36. }
  37. spark.read.format("hudi"). \
  38.     options(**cdc_read_options). \
  39.     load(basePath).show(10, False)

Spark SQL 

  1. -- incrementally query data by path
  2. -- start from earliest available commit, end at latest available commit.
  3. SELECT * FROM hudi_table_changes('path/to/table', 'cdc', 'earliest');
  5. -- start from earliest, end at 202305160000.
  6. SELECT * FROM hudi_table_changes('path/to/table', 'cdc', 'earliest', '202305160000');
  8. -- start from 202305150000, end at 202305160000.
  9. SELECT * FROM hudi_table_changes('path/to/table', 'cdc', '202305150000', '202305160000');

Key requirements Note that CDC queries are currently only supported on Copy-on-Write tables.

Table Types

The examples thus far have showcased one of the two table types, that Hudi supports - Copy-on-Write (COW) tables. Hudi also supports a more advanced write-optimized table type called Merge-on-Read (MOR) tables, that can balance read and write performance in a more flexible manner. See table types for more details.

Any of these examples can be run on a Merge-on-Read table by simply changing the table type to MOR, while creating the table, as below.

Scala

  1. // spark-shell
  2. inserts.write.format("hudi").
  3.   ...
  4.   option("hoodie.datasource.write.table.type", "MERGE_ON_READ").
  5.   ...

Python

  1. # pyspark
  2. hudi_options = {
  3.   ...
  4.   'hoodie.datasource.write.table.type': 'MERGE_ON_READ'
  5. }
  7. inserts.write.format("hudi"). \
  8. options(**hudi_options). \
  9. mode("overwrite"). \
  10. save(basePath)

Spark SQL

  1. CREATE TABLE hudi_table (
  2.     uuid STRING,
  3.     rider STRING,
  4.     driver STRING,
  5.     fare DOUBLE,
  6.     city STRING
  7. ) USING HUDI TBLPROPERTIES (type = 'mor')
  8. PARTITIONED BY (city);

Keys

Hudi also allows users to specify a record key, which will be used to uniquely identify a record within a Hudi table. This is useful and critical to support features like indexing and clustering, which speed up upserts and queries respectively, in a consistent manner. Some of the other benefits of keys are explained in detail here. To this end, Hudi supports a wide range of built-in key generators, that make it easy to generate record keys for a given table. In the absence of a user configured key, Hudi will auto generate record keys, which are highly compressible.

Scala

  1. // spark-shell
  2. inserts.write.format("hudi").
  3. ...
  4. option("hoodie.datasource.write.recordkey.field", "uuid").
  5. ...

Python

  1. # pyspark
  2. hudi_options = {
  3.   ...
  4.   'hoodie.datasource.write.recordkey.field': 'uuid'
  5. }
  7. inserts.write.format("hudi"). \
  8. options(**hudi_options). \
  9. mode("overwrite"). \
  10. save(basePath)

Spark SQL

  1. CREATE TABLE hudi_table (
  2.     ts BIGINT,
  3.     uuid STRING,
  4.     rider STRING,
  5.     driver STRING,
  6.     fare DOUBLE,
  7.     city STRING
  8. ) USING HUDI TBLPROPERTIES (primaryKey = 'uuid')
  9. PARTITIONED BY (city);

Implications of defining record keys Configuring keys for a Hudi table, has a new implications on the table. If record key is set by the user, upsert is chosen as the write operation. Also if a record key is configured, then it’s also advisable to specify a precombine or ordering field, to correctly handle cases where the source data has multiple records with the same key. See section below.

Ordering Field

Hudi also allows users to specify a precombine field, which will be used to order and resolve conflicts between multiple versions of the same record. This is very important for use-cases like applying database CDC logs to a Hudi table, where a given record may be appear multiple times in the source data due to repeated upstream updates. Hudi also uses this mechanism to support out-of-order data arrival into a table, where records may need to be resolved in a different order than their commit time. For e.g using a created_at timestamp field as the precombine field will prevent older versions of a record from overwriting newer ones or being exposed to queries, even if they are written at a later commit time to the table. This is one of the key features, that makes Hudi, best suited for dealing with streaming data.

Scala

  1. // spark-shell 
  2. updatesDf.write.format("hudi").
  3.   ...
  4.   option("hoodie.datasource.write.precombine.field", "ts").
  5.   ...

Python

  1. # pyspark
  2. hudi_options = {
  3. ...
  4. 'hoodie.datasource.write.precombine.field': 'ts'
  5. }
  7. upsert.write.format("hudi").
  8.     options(**hudi_options).
  9.     mode("append").
  10.     save(basePath)

Spark Sql

  1. CREATE TABLE hudi_table (
  2.     ts BIGINT,
  3.     uuid STRING,
  4.     rider STRING,
  5.     driver STRING,
  6.     fare DOUBLE,
  7.     city STRING
  8. ) USING HUDI TBLPROPERTIES (preCombineField = 'ts')
  9. PARTITIONED BY (city);

Where to go from here?

You can also build hudi yourself and try this quickstart using --jars <path to spark bundle jar>(see also build with scala 2.12) for more info. If you are looking for ways to migrate your existing data to Hudi, refer to migration guide.

Spark SQL Reference

For advanced usage of spark SQL, please refer to Spark SQL DDL and Spark SQL DML reference guides. For alter table commands, check out this. Stored procedures provide a lot of powerful capabilities using Hudi SparkSQL to assist with monitoring, managing and operating Hudi tables, please check this out.

Streaming workloads

Hudi provides industry-leading performance and functionality for streaming data.

Hudi Streamer - Hudi provides an incremental ingestion/ETL tool - HoodieStreamer, to assist with ingesting data into Hudi from various different sources in a streaming manner, with powerful built-in capabilities like auto checkpointing, schema enforcement via schema provider, transformation support, automatic table services and so on.

Structured Streaming - Hudi supports Spark Structured Streaming reads and writes as well. Please see here for more.

Check out more information on modeling data in Hudi and different ways to perform batch writes and streaming writes.

Dockerized Demo

Even as we showcased the core capabilities, Hudi supports a lot more advanced functionality that can make it easy to get your transactional data lakes up and running quickly, across a variety query engines like Hive, Flink, Spark, Presto, Trino and much more. We have put together a demo video that showcases all of this on a docker based setup with all dependent systems running locally. We recommend you replicate the same setup and run the demo yourself, by following steps here to get a taste for it.