Connecting Apache Spark with Data Engine

Usage within Watson Studio Notebooks

Watson Studio has the compatible Hive metastore client already included. It also includes a convenience library to configure the connection to the Hive metastore. Set the required variables (CRN and apikey) and call the helper function to connect to the Hive metastore:

# change the CRN and the APIkey according to your instance
crn='yourDataengineCRN'
apikey='yourAPIkey'

from dataengine import SparkSessionWithDataengine

# call the helper function to create a session builder equipped with the correct config
session_builder = SparkSessionWithDataengine.enableDataengine(crn, apikey, "public")
spark = session_builder.appName("Spark DataEngine integration test").getOrCreate()

Display your tables and run an SQL statement:

spark.sql('show tables').show(truncate=False)

# replace yourTable with a valid table. Do not use the sample tables as you do not have access to the data!
spark.sql('select * from yourTable').show()

Usage with IBM Analytics Engine

The Hive metastore client is also already included in IBM® Analytics Engine. It also includes the convenience library to configure the connection to the Hive metastore. The following example shows a Spark batch job for a show tables example in Python:

import sys
from dataengine import SparkSessionWithDataengine

if __name__ == '__main__':
    crn = sys.argv[1]
    apikey = sys.argv[2]

    print(" Start SparkSessionWithDataengine example")
    session_builder = SparkSessionWithDataengine.enableDataengine(crn, apikey, "public")

    print(" Setup IBM Cloud Object Storage access")
    spark = session_builder.appName("AnalyticEngine DataEngine integration") \
          .config("fs.cos.impl", "com.ibm.stocator.fs.ObjectStoreFileSystem") \
          .config("fs.stocator.scheme.list", "cos") \
          .config("fs.stocator.cos.impl", "com.ibm.stocator.fs.cos.COSAPIClient") \
          .getOrCreate()

    print(" Got a spark session, listing all tables")
    spark.sql('show tables').show()

    spark.stop()

Prepare a JSON file to start that program, as in the following example (listTablesExample.json):

{
  "application_details": {
     "application": "cos://<your-bucket>.<cos-region>/listTablesExample.py",
     "arguments": ["<Data-Engine-instance-CRN>", "<API-key-to-access-data-engine-instance>"],
     "conf": {
        "ae.spark.executor.count":"1",
        "ae.spark.autoscale.enable":"false",
        "spark.hadoop.fs.cos.<cos-region>.endpoint": "https://s3.direct.us-south.cloud-object-storage.appdomain.cloud",
        "spark.hadoop.fs.cos.<cos-region>.iam.api.key": "<API-key-to-access-python-file>",
        "spark.app.name": "DataEngineHiveAccess"
     }
  }
}

Start the application by using a curl command, as in the following example:

curl -X POST https://api.us-south.ae.cloud.ibm.com/v3/analytics_engines/<GUID of Analytic Engine>/spark_applications --header "Authorization: Bearer $TOKEN" -H "content-type: application/json"  -d @listTablesExample.json

Apache Spark Data Engine integration

For self-hosted Apache Spark installations use the following instructions.

1. Ensure that Stocator is installed according to the instructions provided.

2. Download the Hive-compatible client with the provided instructions.

3. Either download the provided convenience libraries or, in case you don't want to use them, set the following settings in SparkContext yourself:

   spark = SparkSession.builder.appName('Python-App') \
       .config("spark.sql.pyspark.jvmStacktrace.enabled", True) \
       .config("spark.hive.metastore.truststore.path", "file:///opt/ibm/jdk/jre/lib/security/cacerts") \
       # to access IBM Cloud Object Storage ensure that stocator is available
       .config("fs.cos.impl", "com.ibm.stocator.fs.ObjectStoreFileSystem") \
       .config("fs.stocator.scheme.list", "cos") \
       .config("fs.stocator.cos.impl", "com.ibm.stocator.fs.cos.COSAPIClient") \
       .config("fs.stocator.cos.scheme", "cos") \
       # register the required Cloud Object Storage path used in our application, add endpoints for all buckets
       .config("spark.hadoop.fs.cos.us-geo.endpoint", "https://s3.us.cloud-object-storage.appdomain.cloud") \
       .config("spark.hadoop.fs.cos.us-geo.iam.endpoint", "https://iam.cloud.ibm.com/identity/token") \
       .config("spark.hadoop.fs.cos.us-geo.iam.api.key", '<YourAPIkey>') \
       .config("spark.sql.hive.metastore.version", "3.0") \
       # directory where the Hive client has been placed
       .config("spark.sql.hive.metastore.jars", "/tmp/dataengine/*") \
       .config("spark.hive.metastore.uris", "thrift://catalog.<region>.sql-query.cloud.ibm.com:9083") \
       .config("spark.hive.metastore.use.SSL", "true") \
       .config("spark.hive.metastore.truststore.password", "changeit") \
       .config("spark.hive.metastore.client.auth.mode", "PLAIN") \
       .config("spark.hive.metastore.client.plain.username", '<YourDataengineCRN>') \
       .config("spark.hive.metastore.client.plain.password", '<YourAPIkey>') \
       .config("spark.hive.execution.engine", "spark") \
       .config("spark.hive.stats.autogather", "false") \
       .config("spark.sql.warehouse.dir", "file:///tmp") \
       # only spark is allowed as the default catalog
       .config("metastore.catalog.default", "spark") \
       .enableHiveSupport() \
       .getOrCreate()
   

package com.ibm.cloud.dataengine
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession.{Builder => SessionBuilder}
import SparkSessionBuilderAddOn._
object SparkSessionBuilderHMSConfigTest {
  def main(args: Array[String]) = {
    val spark = SparkSession
      .builder()
      .appName("Spark DataEngine integration")
      .enableDataengine(args(0), args(1), "public")
      .config("fs.cos.impl", "com.ibm.stocator.fs.ObjectStoreFileSystem")
      .config("fs.stocator.scheme.list", "cos")
      .config("fs.stocator.cos.impl", "com.ibm.stocator.fs.cos.COSAPIClient")
      .config("fs.stocator.cos.scheme", "cos")
      .getOrCreate()
    println("Got a spark session, listing all tables")
    val sqlDf = spark.sql("SHOW TABLES")
    sqlDf.show()
  }
}

Troubleshooting error message

If you receive the following error message when you run the SQL statement, check which of the possible causes exist.

AnalysisException: org.apache.hadoop.hive.ql.metadata.HiveException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient

The following possible causes might exist.

• The CRN is invalid or the service does not exist.
• THE APIKEY is invalid.
• The Data Engine service has a Lite plan.

Py4JJavaError: An error occurred while calling o8820.sql.
: java.util.concurrent.ExecutionException: com.ibm.stocator.fs.common.exception.ConfigurationParseException: Configuration parse exception: Access KEY is empty. Please provide valid access key

The following possible cause might exist.

• Tables that are based on sample data that is provided by Data Engine do not work for a SELECT statement. Therefore, use data stored in your Cloud Object Storage bucket.