Submitting Spark application by using REST API

You can submit a Spark application by running a CURL command. Complete the following steps to submit a Python application.

watsonx.data on IBM Software Hub

Procedure

Create a storage volume to store the Spark application and related output.
- Option1: Create a storage volume in IBM Software Hub. To create storage volume in IBM Software Hub, see Creating a storage volume.
- Option2: Create Cloud Object Storage. To create Cloud Object Storage and a bucket, see Creating a storage bucket.
If you use Cloud Object Storage, register the Cloud Object Storage in watsonx.data, register Cloud Object Storage bucket. To register a Cloud Object Storage bucket, see Adding storage.
Upload the Spark application to the storage volume.
- If you use IBM Software Hub storage volume, see Creating a storage volume.
- If you use Cloud Object Storage, see Uploading data.
If your Spark application resides in IBM Software Hub storage volume, specify the parameter values and run the following CURL command to submit the application.
```
curl --request POST \
  --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'LhInstanceId: <instance_id>' \
  --data '{
  "application_details": {
    "application": "/myapp/<python file name>"
  },
  "volumes": [
    {
      "name": "cpd-instance::my-vol-1",
      "mount_path": "/myapp"
    }
  ]
}'
```
Parameter values:
- <cpd_host_name>: The hostname of your IBM Software Hub.
- <api_version>: When using the v2 API, set the <api_version> parameter to v2; for the v3 API, set it to v3.
- <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, 1609968577169454.
- <python file name> : The Spark application file name. It must be available in the storage volume.
- <my-vol-1> : The display the name of the storage volume.
Submitting an application by accessing the watsonx.data catalog.
Use the following command to access data from a catalog that is associated with the Spark engine and to perform some basic operations on that catalog.
Example:
```
curl --request POST \
  --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'LhInstanceId: <instance_id>' \
  --data '{
  "application_details": {
    "application": "s3a://<application-bucket-name>/iceberg.py",
    "conf": {
        "spark.hadoop.wxd.apiKey":"ZenApiKey <encoded key>",
        "spark.app.name": "reader-app"
    }
  }
}'
```
Parameter values:
- <cpd_host_name>: The hostname of your IBM Software Hub.
- <api_version>: When using the v2 API, set the <api_version> parameter to v2; for the v3 API, set it to v3.
- <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.
- <application-bucket-name> : The display the name of the storage volume.
Important: For Apache Gluten accelerated Spark engine, if your Spark application resides in Amazon S3 storage, you can use DAS to submit your application.
If your Spark application resides in ADLS(Gen1 or Gen2) and you want to submit the application by using DAS, specify the parameter values and run the following curl command. The following example shows the command to submit read.py application.

Important: Azure Data Lake Storage (ADLS) Gen1 is deprecated and will be removed in an upcoming release. You must transition to ADLS Gen2 as ADLS Gen1 will no longer be available.
Example 1:
```
curl --request POST \
  --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'LhInstanceId: <instance_id>' \
  --data '{
  "application_details": {
    "application": "abfss://<storage_account>@<storage_container>.dfs.core.windows.net/adls-read.py",
    "conf": {
        "spark.hadoop.wxd.apikey":<token>,
        "spark.app.name": "reader-app"
    }
  }
}'
```
Parameter values:
- <cpd_host_name>: The hostname of your IBM Software Hub.
- <api_version>: When using the v2 API, set the <api_version> parameter to v2; for the v3 API, set it to v3.
- <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.
- <storage_account> : The name of the azure storage account.
- <storage_container> : The name of the Azure storage container.

If your Spark application resides in Google Cloud Storage and you want to submit the application by using DAS, specify the parameter values and run the following curl command. The following example shows the command to submit read.py application.

Example 1 :

curl --request POST \
  --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --header 'LhInstanceId: <instance_id>' \
  --data '{
  "application_details": {
    "application": "gs://<application-bucket-name>//gcs-read.py",
    "conf": {
        "spark.hadoop.wxd.apikey":<token>,
        "spark.app.name": "reader-app"
    }
  }
}

Parameter values:

<cpd_host_name>: The hostname of your watsonx.data cluster.
<api_version>: When using the v2 API, set the <api_version> parameter to v2; for the v3 API, set it to v3.
<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.
<application-bucket-name> : The display the name of the storage volume.

If you are using Apache Gluten accelerated Spark engine, for ADLS, use the following command:

  {
 "application_details": {
   "conf": {
       "spark.hadoop.fs.azure.account.auth.type.<storage_account>.dfs.core.windows.net": "OAuth",
       "spark.hadoop.fs.azure.account.oauth.provider.type.<storage_account>.dfs.core.windows.net": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
       "spark.hadoop.fs.azure.account.oauth2.client.id.<storage_account>.dfs.core.windows.net": "<application-id>",
       "spark.hadoop.fs.azure.account.oauth2.client.secret.<storage_account>.dfs.core.windows.net":"<secret-id>",
       "spark.hadoop.fs.azure.account.oauth2.client.endpoint.<storage_account>.dfs.core.windows.net": "https://login.microsoftonline.com/<directory-id>/oauth2/token",
       "spark.hadoop.wxd.apikey": "ZenApiKey xxx"
   },

   "application": "abfss://<storage_account>@<storage_container>.dfs.core.windows.net/adls-read.py"
 }
}

If you are using Apache Gluten accelerated Spark engine, for GCS, use the following command:

         {
               "application_details": {
                 "conf": {

                     "ae.spark.driver.log.level":"INFO",
                     "ae.spark.executor.log.level":"INFO",
                     "spark.executor.cores": "2",
                     "spark.executor.memory": "4G",
                     "spark.driver.cores": "2",
                     "spark.driver.memory": "4G",
                     "ae.spark.executor.count": "1",
                     "spark.hadoop.fs.gs.auth.type": "SERVICE_ACCOUNT_JSON_KEYFILE",
                     "spark.hadoop.fs.gs.auth.service.account.json.keyfile": "/mnts/test/<account-key-file-name>.json"
                     "spark.hadoop.wxd.apikey": "ZenApiKey xxx"
                 },
                  "application": "gs://<application-bucket-name>//gcs-read.py"
               },
                 "volumes": [
                 {
                   "name": "cpd-instance::my-vol",
                   "mount_path": "/mnts/test"
                 }
               ]
             }

Submitting an application by accessing the Amazon S3 Multi-Region Access Point (MRAP) storage. In this case, you must specify the following additional configuration:

.config("spark.hadoop.fs.s3a.bucket.m6edq7uym8kk7.mrap.endpoint.region", "us-east-1")

Sample configuration for using the Hadoop Iceberg catalog:



def init_spark():
    spark = (
        SparkSession.builder
        .appName("Iceberg-MRAP-Hadoop")
        .config(f"spark.sql.catalog.{catalog_name}.type", "hadoop")
        .config(f"spark.sql.catalog.{catalog_name}.warehouse", warehouse_path)
        .config("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider")
        .config("spark.hadoop.fs.s3a.access.key", "AKIAYOHCHY5F5LUUJ6XT")
        .config("spark.hadoop.fs.s3a.secret.key", "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")
        
        .getOrCreate()
    )

    # spark.sparkContext.setLogLevel("DEBUG")
    return spark

After you submit the Spark application, you receive a confirmation message with the application ID and Spark version. Save it for reference.
Log in to the watsonx.data cluster, access the Engine details page. In the Applications tab, use the application ID to list the application and you can track the stages. For more information, see View and manage applications.
Note: When you try to submit a Spark application by using API and if the JSON payload includes an error, the job fails without generating logs. . To troubleshoot, see Troubleshooting section.
1. By using the Spark Jobs API:
  - List all active jobs:
```
curl -k -X GET --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications?state=RUNNING -H "Authorization: ZenApiKey ${TOKEN}"
```
    Note that jobs in FINISHED state are removed and don't show up in the list of all jobs.
  - Get the status of a job:
```
curl -k -X GET --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications/<job_id> -H "Authorization: ZenApiKey ${TOKEN}"
```
    Example response:
```
{ "application_id": "28ce7f71-a357-4583-9de8-6607047ca783", "state": "RUNNING", "start_time": "Monday' 07 June 2021 '14:46:23.237+0000", "spark_application_id": "app-20210607144623-0000" }
```
  - List all jobs:
```
curl -k -X GET --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications?state=RUNNING -H "Authorization: ZenApiKey ${TOKEN}"
```
  - List all jobs with certain states:
    
    The following example lists all jobs in RUNNING or FAILED state. The different job states that you can query through the API are listed in the table at the beginning of this section.
```
curl -k -X GET --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications?state=WAITING,RUNNING,FAILED,UNKNOWN,STOPPED
```
Stopping Spark jobs
You can stop a Spark job by using the Spark Jobs API:
- curl -k -X DELETE --url https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications/<job-id> -H "Authorization: ZenApiKey ${TOKEN}"
  
  Returns 204 No Content if the job was successfully deleted.
  
  If Running jobs are stopped, the Spark application will appear under the Incomplete Applications tab in Spark history server.