Access data with Spark

When using a Domino on-demand Spark cluster, any data that will be used, created, or modified as part of the interaction must go into an external data store.

Note	On-demand Spark clusters are not intended as a permanent store of any data or collocating a big data layer such as HDFS. Any data that is not stored externally from the cluster will be lost upon termination.

Use Domino datasets

When you create a Spark cluster attached to a Domino workspace or job, any Domino dataset accessible from the workspace or job will also be accessible from all components of the cluster under the same dataset mount path. Data can be accessed using the following path prefix:

file:///

For example, to read a file you would use the following.

rdd = sc.textFile("file:///path/to/file")

No additional configuration of the Spark cluster environment or the execution environment is required.

Use S3

To enable working with data in Amazon S3 (or S3 compatible object store) you must ensure that your base Spark cluster environment and compatible PySpark compute environment are configured with the Hadoop-AWS module.

The environments created when configuring prerequisites will at a minimum include Hadoop 2.7.3 client libraries, which are sufficient for basic access. A number of additional commonly used features (for example, temporary credentials, SSE-KMS encryption, more efficient committers, etc.) are only available in more recent Hadoop-AWS module versions.

Consult the documentation for the relevant version to determine what may be the best fit for you.

For Spark 2.4.x, a good advanced option would be Hadoop 2.9.2.

S3 usage examples

Now that you have your environments properly setup, you can interact with S3. The following are several common access patterns.

Access bucket with AWS credentials in environment variables

import os
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# the default configuration will pick up your credentials from environment variables
# No additional configuration is necessary

# test reading
df = spark.read.json("s3a://bucket/prefix1/prefix2/people.json")
df.show()

Access bucket with SSE-KMS encryption

Note	Requires Hadoop-AWS 2.9.3+

import os
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# for write operations you will need the ARN of the key to use
# Note that the credentials used need to have proper access to use the key
kms_key_arn = "<your key ARN here>"

# configure the connector
# This example assumes credentials from environment variables so no need to configure
# Note: The encryption config is not needed for read only operations
hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
hadoop_conf.set("fs.s3a.server-side-encryption-algorithm", "SSE-KMS")
hadoop_conf.set("fs.s3a.server-side-encryption.key", kms_key_arn)

# test reading
df = spark.read.json("s3a://bucket/prefix1/prefix2/people.json")
df.show()

# test writing
df.write.mode("overwrite").parquet("3a://bucket/prefix1/prefix2/write-test/output")

Access a bucket with Domino assumed temporary credentials

Note	Requires Hadoop-AWS 2.9.3+ It is important that no AWS credential variables are set in your user profile or project.

import os
from pyspark.sql import SparkSession

try:
    spark.stop()
except:
    pass
spark = SparkSession.builder.getOrCreate()

#The name of one of the roles you are entitled to
profile_name="my-role-name-read-write"

# use boto3 for convenience to get credentials form credentials file populated by Domino
# can use any method desirable to extract the credentials
import boto3
role_creds = boto3.Session(profile_name=profile_name).get_credentials().get_frozen_credentials()

# configure the connector
# Use the TemporaryAWSCredentialsProvider
hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
hadoop_conf.set("fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider")
hadoop_conf.set("fs.s3a.access.key", role_creds.access_key)
hadoop_conf.set("fs.s3a.secret.key", role_creds.secret_key)
hadoop_conf.set("fs.s3a.session.token", role_creds.token)

# test reading
df = spark.read.json("s3a://bucket/prefix1/prefix2/people.json")
df.show()

# test writing
df.write.mode("overwrite").parquet("s3a://bucket/prefix1/prefix2/write-test/output")

For a full set of configuration options, see the documentation for the Hadoop-AWS module.

Using Azure Data Lake Storage Gen2

To enable working with data in Azure Data Lake Storage (ADLS) Gen2 you need to configure your base Spark environment and your compute environment with the Hadoop-Azure ABFS connector.

The ABFS connector requires Hadoop 3.2+.

To accomplish this, set SPARK_VERSION=3.0.0 and HADOOP_VERSION=3.2.1 when following the advanced instructions for base Spark cluster environment and compatible PySpark compute environment.

Note	You must also enable the `ENV HADOOP_OPTIONAL_TOOLS=hadoop-azure` directive in your environments.