Working with Apache Hudi catalog

The topic describes the procedure to run a Spark application that ingests data into an Apache Hudi catalog.

Before you begin

To enable your Spark application to work with the watsonx.data catalog and storage, you must have Metastore admin role. Without Metastore admin privilege, you cannot ingest data to storage using Native Spark engine. To enable your Spark application to work with the watsonx.data catalog and storage, add the following configuration to your application payload:.
spark.hadoop.wxd.apiKey=Basic base64(ibmlhapikey_ibmcloudid:apikey)

Procedure

  1. Create a storage with Apache Hudi catalog to store data used in the Spark application. To create storage with Apache Hudi catalog, see Adding a storage-catalog pair.
  2. Associate the storage with the Native Spark engine. For more information, see Associating a catalog with an engine.
  3. Create Cloud Object Storage (COS) to store the Spark application. To create Cloud Object Storage and a bucket, see Creating a storage bucket.
  4. Register the Cloud Object Storage in watsonx.data. For more information, see Adding a storage-catalog pair.
  5. Save the following Spark application (Python file) to your local machine. Here, hudi_demo.py.
    The Python Spark application demonstrates the following functionality:
    • It creates a database inside the Apache Hudi catalog (that you created to store data). Here, hudi_db.
    • It creates a table inside the hudi_db database, namely hudi_table.
    • It inserts data into the hudi_table and does SELECT query operation.
    • It drops the table and schema after use. 
    from pyspark.sql import SparkSession

def init_spark():
    spark = SparkSession.builder \
        .appName("CreateHudiTableInCOS") \
        .enableHiveSupport() \
        .getOrCreate()
    return spark

def main():

    try:
        spark = init_spark()
        spark.sql("show databases").show()
        spark.sql("create database if not exists spark_catalog.hudi_db LOCATION 's3a://hudi-connector-test/'").show()
        spark.sql("create table if not exists spark_catalog.hudi_db.hudi_table (id bigint, name string, location string) USING HUDI OPTIONS ('primaryKey' 'id', hoodie.write.markers.type= 'direct', hoodie.embed.timeline.server= 'false')").show()
        spark.sql("insert into hudi_db.hudi_table VALUES (1, 'Sam','Kochi'), (2, 'Tom','Bangalore'), (3, 'Bob','Chennai'), (4, 'Alex','Bangalore')").show()
        spark.sql("select * from spark_catalog.hudi_db.hudi_table").show()
        spark.sql("drop table spark_catalog.hudi_db.hudi_table").show()
        spark.sql("drop schema spark_catalog.hudi_db CASCADE").show()

    finally:
        spark.stop()

if __name__ == '__main__':
    main()
  6. Upload the Spark application to the COS, see Uploading data.
  7. To submit the Spark application with data residing in Cloud Object Storage, specify the parameter values and run the following curl command. 
    curl --request POST \
  --url https://<wxd_host_name>/lakehouse/api/v2/spark_engines/<spark_engine_id>/applications \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'LhInstanceId: <instance_id>' \
  --data '{
	"application_details": {
		"conf": {
        "spark.sql.catalog.spark_catalog.type": "hive",
        "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog",
        "spark.hadoop.wxd.apiKey":"ZenApiKey <user-authentication-string> "

		},
		"application": "s3a://hudi-connector-test/hudi_demo.py"
	}
}
    Parameter values:
    • <wxd_host_name>: The hostname of your watsonx.data.
    • <spark_engine_id> : The Engine ID of the native Spark engine.
    • <token> : The bearer token. For more information about generating the token, see Generating a bearer token.
    • <instance_id> : The instance ID from the watsonx.data cluster instance URL. For example, 1609968977179454.
    • <user-authentication-string> : The value must be in the format : echo -n "<username>:<your Zen API key>" | base64. The Zen API Key here is the API key of the user accessing the Object store bucket. To generate API key, log in into the watsonx.data console and navigate to Profile > Profile and Settings > API Keys and generate a new API key.
      Note: If you generate a new API key, your old API key becomes invalid.
  8. After you submit the Spark application, you receive a confirmation message with the application ID and Spark version. Save it for reference.
  9. Log in to the watsonx.data cluster, access the Engine details page. In the Applications tab, use application ID to list the application and track the stages. For more information, see View and manage applications.