Connect PySpark with the Data Platform Python SDK#

#Objective

Do not hesitate to check out the sample code for common use cases.

#The connect module with PySpark

When you create a Custom PySpark action, two things are done by Data Platform:

• SparkContext is created before calling the script.
• SparkContext.stop() is called after the script.

So you can simply access Spark's functionalities with:

SparkContext.getOrCreate()

To access the data stored within Data Platform, you have to first instantiate a Connector object with the connect function. Then you can use Spark-compatible methods in the Connector object to interact with the connected data.

For instance, the method for getting a Spark Dataframe object is connnector.extract_dataframe(). The sample code below connects directly to a source and then uses that method to get the Spark object:

from forepaas.dwh import connect

cn_source = connect("dwh/file_upload_source/chicago_calendar_full.csv")

# Spark compatible connector extract_dataframe function returns Spark DataFrame
spark_df = cn_source.extract_dataframe()

#Spark-compatible Connector Methods

#get_spark_options()

Returns Spark options to connect to a Database.

Not available with Protocol sources (Buckets, File Upload, Amazon S3, etc).

Output

{
            "sfAccount": ...,
            "sfURL": ...,
            "sfUser": ...,
            "sfSchema": ...,
            "sfDatabase": ...,
            "sfTimezone": ...,
            "sfWarehouse": ...,
            "preactions": ...,
}

#get_spark_context()

Returns the current SparkContext.

Output

#get_spark_session()

Returns current spark session, configured with different settings when using Data Platform bucket / file-upload.

Available only with Protocol sources (Buckets, File Upload, Amazon S3, etc).

Output

#get_spark_url(path, filename, bucket=None)

Returns file url to given path of object in object store. path needs to be absolute path in object store, and if bucket is not set, it uses the one from the user config

Available only with Protocol sources (Buckets, File Upload, Amazon S3, etc).

Input Parameters

Output

#insert_dataframe(table, dataframe)

Inserts a Spark DataFrame into the connected source.

Available with Databases and some Protocol sources (Buckets, Amazon S3, Azure Blob Storage). For more details, check the use case 4 and 5.

Input Parameters

#extract_dataframe(params={})

Extract SparkDataframe from file. File extraction options can be set via params or in table.parameters

Input Parameters

Detailed example:

 params = {
                "scale": ["attribute_1", "attribute_2"],
                "joins": [{
                    "type": "INNER",
                    "table": "table_1",
                    "condition": "table_1.attribute = table_2.attribute"
                }],
                "schema": {
                    "attribute_dest": "attribute_source"
                }
            }

Output

#Go further

If you need training or technical assistance to implement our solutions, contact your sales representative or click on this link to get a quote and ask our Professional Services experts for a custom analysis of your project.

Ask questions, give your feedback and interact directly with the team building the Data Platform on the dedicated Discord channel.

If you need support with your OVHcloud services, create a request in our Help Centre.

Join our community of users.

¹: S3 is a trademark of Amazon Technologies, Inc. OVHcloud's service is not sponsored by, endorsed by, or otherwise affiliated with Amazon Technologies, Inc.