Querying Cloudera tables using Spark engine

You can query Hive and Iceberg tables stored in Cloudera HDFS from watsonx.data using the Spark engine with Kerberos authentication. This integration provides read-only, zero-copy access to remote Cloudera data through PySpark applications.

1. Configure the Spark engine with a volume mount and object storage.
   1. Log in to IBM Software Hub.
   2. Click Navigation Menu > Storage volumes.
   3. Click New volume and provide the following details:
      • Volume name: Enter a name (for example, cloudera-conf-vol).
      • Type: Select New PVC.
      • Storage class: Select nfs-client.
      • Size: Enter the required storage size.
      • Mount path: Enter the mount path (for example, /mnts/cloudera).
   4. Click Add.
   5. Navigate to the created volume and click File browser.
   6. Upload the following Cloudera configuration files to the volume:
      • core-site.xml
      • hdfs-site.xml
      • hive-site.xml
      • krb5.conf
      • Your Kerberos keytab file (for example, hive.keytab)
   7. In the watsonx.data console, navigate to Infrastructure manager and select your Spark engine.
   8. Click Edit and configure the object storage bucket:
      • Bucket name: Enter your bucket name.
      • Endpoint: Enter the bucket endpoint URL.
      • Access key and Secret key: Enter the bucket access credentials.
   9. Under Volume mounts, associate the volume created in the step with the Spark engine and enter its mount path.
   10. Click Save.
2. Prepare the PySpark application script.

   Create a PySpark script and upload it to your object storage bucket. Choose the script that matches your table type.

   For Hive tables:

   from pyspark.sql import SparkSession
   from pyspark import SparkConf
   import subprocess
   
   # Kerberos and HDFS configuration
   KEYTAB_PATH      = "<mount_path>/<keytab_file>"
   PRINCIPAL        = "<kerberos_principal>"
   HIVE_METASTORE   = "thrift://<metastore_host>:<metastore_port>"
   NAMENODE_URI     = "hdfs://<namenode_host>:<namenode_port>"
   KRB5_CONF        = "<mount_path>/krb5.conf"
   CORE_SITE        = "<mount_path>/core-site.xml"
   HDFS_SITE        = "<mount_path>/hdfs-site.xml"
   HIVE_SITE        = "<mount_path>/hive-site.xml"
   DATABASE_NAME    = "<database_name>"
   TABLE_NAME       = "<table_name>"
   
   # Obtain a Kerberos ticket
   subprocess.run(
       ["kinit", "-kt", KEYTAB_PATH, PRINCIPAL],
       check=True
   )
   print(f"Kerberos ticket obtained for: {PRINCIPAL}")
   
   conf = SparkConf()
   conf.set("spark.hadoop.fs.defaultFS", NAMENODE_URI)
   conf.set("spark.hadoop.hadoop.security.authentication", "kerberos")
   conf.set("spark.hadoop.hadoop.security.authorization", "true")
   conf.set("spark.hadoop.dfs.namenode.kerberos.principal", PRINCIPAL)
   conf.set("spark.kerberos.keytab", KEYTAB_PATH)
   conf.set("spark.kerberos.principal", PRINCIPAL)
   conf.set("spark.hadoop.hadoop.kerberos.kinit.command", "kinit")
   conf.set("spark.hadoop.java.security.krb5.conf", KRB5_CONF)
   
   spark = SparkSession.builder \
       .appName("ClouDeraHiveQuery") \
       .config(conf=conf) \
       .enableHiveSupport() \
       .getOrCreate()
   
   spark.sparkContext.addFile(CORE_SITE)
   spark.sparkContext.addFile(HDFS_SITE)
   spark.sparkContext.addFile(HIVE_SITE)
   
   spark.sql(f"USE {DATABASE_NAME}")
   df = spark.sql(f"SELECT * FROM {TABLE_NAME} LIMIT 20")
   df.printSchema()
   df.show(20, truncate=False)
   print(f"Total rows: {df.count()}")
   spark.stop()

   For Iceberg tables:

   from pyspark.sql import SparkSession
   from pyspark import SparkConf
   import subprocess
   
   # Kerberos and HDFS configuration
   KEYTAB_PATH      = "<mount_path>/<keytab_file>"
   PRINCIPAL        = "<kerberos_principal>"
   HIVE_METASTORE   = "thrift://<metastore_host>:<metastore_port>"
   NAMENODE_URI     = "hdfs://<namenode_host>:<namenode_port>"
   KRB5_CONF        = "<mount_path>/krb5.conf"
   CORE_SITE        = "<mount_path>/core-site.xml"
   HDFS_SITE        = "<mount_path>/hdfs-site.xml"
   HIVE_SITE        = "<mount_path>/hive-site.xml"
   CATALOG_NAME     = "<iceberg_catalog_name>"
   DATABASE_NAME    = "<database_name>"
   TABLE_NAME       = "<table_name>"
   
   # Obtain a Kerberos ticket
   subprocess.run(
       ["kinit", "-kt", KEYTAB_PATH, PRINCIPAL],
       check=True
   )
   print(f"Kerberos ticket obtained for: {PRINCIPAL}")
   
   conf = SparkConf()
   conf.set("spark.hadoop.fs.defaultFS", NAMENODE_URI)
   conf.set("spark.hadoop.hadoop.security.authentication", "kerberos")
   conf.set("spark.hadoop.hadoop.security.authorization", "true")
   conf.set("spark.hadoop.dfs.namenode.kerberos.principal", PRINCIPAL)
   conf.set("spark.kerberos.keytab", KEYTAB_PATH)
   conf.set("spark.kerberos.principal", PRINCIPAL)
   conf.set("spark.hadoop.hadoop.kerberos.kinit.command", "kinit")
   conf.set("spark.hadoop.java.security.krb5.conf", KRB5_CONF)
   conf.set(f"spark.sql.catalog.{CATALOG_NAME}",
       "org.apache.iceberg.spark.SparkCatalog")
   conf.set(f"spark.sql.catalog.{CATALOG_NAME}.type", "hive")
   conf.set(f"spark.sql.catalog.{CATALOG_NAME}.uri", HIVE_METASTORE)
   conf.set("spark.sql.extensions",
       "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
   
   spark = SparkSession.builder \
       .appName("ClouderaIcebergQuery") \
       .config(conf=conf) \
       .getOrCreate()
   
   spark.sparkContext.addFile(CORE_SITE)
   spark.sparkContext.addFile(HDFS_SITE)
   spark.sparkContext.addFile(HIVE_SITE)
   
   df = spark.sql(
       f"SELECT * FROM {CATALOG_NAME}.{DATABASE_NAME}.{TABLE_NAME} LIMIT 20"
   )
   df.printSchema()
   df.show(20, truncate=False)
   print(f"Total rows: {df.count()}")
   
   # Example: Iceberg time travel
   df_snapshot = spark.sql(
       f"SELECT * FROM {CATALOG_NAME}.{DATABASE_NAME}.{TABLE_NAME} "
       f"VERSION AS OF 1"
   )
   df_snapshot.show(truncate=False)
   spark.stop()

   Replace the following placeholders in the script:

   Placeholder	Description
   <mount_path>	Mount path of the storage volume configured in step 1 (for example, /mnts/cloudera).
   <keytab_file>	File name of the Kerberos keytab (for example, hive.keytab).
   <kerberos_principal>	Kerberos principal in the format <service>/<hostname>@<REALM> (for example, hive/cloudera-host.example.com@EXAMPLE.COM).
   <metastore_host>	Hostname or IP address of the Hive Metastore server.
   <metastore_port>	Port of the Hive Metastore (default: 9083).
   <namenode_host>	Hostname or IP address of the HDFS NameNode.
   <namenode_port>	Port of the HDFS NameNode (default: 8020).
   <database_name>	Name of the Hive database containing the target table.
   <table_name>	Name of the Hive or Iceberg table to query.
   <iceberg_catalog_name>	Name for the Iceberg catalog configured in Spark (Iceberg script only). This is a logical name used in Spark SQL; for example, cloudera_iceberg.

   After completing the script, upload it to your object storage bucket (for example, s3a://<bucket_name>/cloudera_query.py).

3. Submit the Spark application using the REST API.
   1. Obtain a bearer token for the watsonx.data REST API.
   2. Identify your Spark engine ID from Infrastructure manager.
   3. Submit the application by sending the following POST request:

      curl -X POST \
        "https://<wxdata_host>/lakehouse/api/v1/spark_engines/<spark_engine_id>/applications" \
        -H "Authorization: Bearer <bearer_token>" \
        -H "Content-Type: application/json" \
        -d '{
          "application_details": {
            "application": "s3a://<bucket_name>/<script_name>.py",
            "conf": {
              "spark.hadoop.fs.s3a.bucket.<bucket_name>.endpoint": "<bucket_endpoint>",
              "spark.hadoop.fs.s3a.bucket.<bucket_name>.access.key": "<access_key>",
              "spark.hadoop.fs.s3a.bucket.<bucket_name>.secret.key": "<secret_key>",
              "spark.hadoop.fs.defaultFS": "hdfs://<namenode_host>:<namenode_port>",
              "spark.hadoop.hadoop.security.authentication": "kerberos",
              "spark.hadoop.hadoop.security.authorization": "true",
              "spark.kerberos.keytab": "<mount_path>/<keytab_file>",
              "spark.kerberos.principal": "<kerberos_principal>",
              "spark.hadoop.java.security.krb5.conf": "<mount_path>/krb5.conf",
              "spark.hadoop.dfs.namenode.kerberos.principal": "<kerberos_principal>",
              "spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
              "spark.jars.packages": "org.apache.iceberg:iceberg-spark-runtime-4.0_2.13:1.9.0"
            },
            "volumes": [
              {
                "mount_path": "<mount_path>",
                "name": "<volume_name>"
              }
            ]
          }
        }'

      Replace the following placeholders in the request:

      Placeholder	Description
      <wxdata_host>	Hostname of your watsonx.data instance.
      <spark_engine_id>	ID of the Spark engine, available in Infrastructure manager.
      <bearer_token>	Valid bearer token for watsonx.data REST API authentication.
      <bucket_name>	Name of the object storage bucket where the PySpark script is uploaded.
      <script_name>	File name of the PySpark script (without the .py extension in the path above).
      <bucket_endpoint>	Bucket endpoint URL without the https:// prefix.
      <access_key>	Access key for the object storage bucket.
      <secret_key>	Secret key for the object storage bucket.
      <namenode_host>	Hostname or IP address of the HDFS NameNode.
      <namenode_port>	Port of the HDFS NameNode (default: 8020).
      <mount_path>	Mount path of the storage volume (for example, /mnts/cloudera).
      <keytab_file>	File name of the Kerberos keytab file.
      <kerberos_principal>	Kerberos principal in the format <service>/<hostname>@<REALM>.
      <volume_name>	Name of the storage volume created in step 1.

      Tip: For Hive-only queries (no Iceberg), you can omit the spark.jars.packages and spark.sql.extensions properties.

      A successful submission returns a JSON response containing the application ID:

      {"id":"<application_id>","state":"starting"}

      Note the <application_id> value for monitoring in the next step.

4. Monitor the application status.
   1. Poll the application status using the following GET request:

      curl -X GET \
        "https://<wxdata_host>/lakehouse/api/v1/spark_engines/<spark_engine_id>/applications/<application_id>" \
        -H "Authorization: Bearer <bearer_token>"

      The response contains a state field. Expected progression:

      starting → running → finished

      A state value of finished indicates successful completion. A value of failed indicates an error — see the troubleshooting section for guidance.

5. Verify query results.
   1. In the watsonx.data console, navigate to Infrastructure manager > Spark engine > Applications.
   2. Click the application entry to open its details.
   3. Click View logs to review the application output and confirm that query results are printed as expected.

root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)
 |-- department: string (nullable = true)
 |-- salary: decimal(10,2) (nullable = true)

+---+-----------+----------+---------+
|id |name       |department|salary   |
+---+-----------+----------+---------+
|1  |John Doe   |IT        |75000.00 |
|2  |Jane Smith |HR        |65000.00 |
|3  |Bob Johnson|Finance   |80000.00 |
+---+-----------+----------+---------+