PySpark with AWS

Spark tutorial · PySpark.in

PySpark with AWS

This tutorial shows how to use PySpark with AWS services, from simple tasks (reading/writing S3) to advanced topics (Glue and EMR integration). Each section starts with the basics and then covers intermediate and advanced patterns. Code examples and best practices are included throughout.

Reading from S3

Before reading from S3, ensure Spark can access AWS. For example, set your AWS credentials (via environment variables, IAM roles, or ~/.aws/credentials). Then build a Spark session with the Hadoop AWS libraries. For instance:

```

hide

import os

from pyspark.sql import SparkSession

# Option 1: use environment variables (or IAM role on AWS)

os.environ['AWS_ACCESS_KEY_ID'] = '<YOUR_ACCESS_KEY>'

os.environ['AWS_SECRET_ACCESS_KEY'] = '<YOUR_SECRET_KEY>'

spark = SparkSession.builder \

.appName("ReadFromS3") \

.config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.1") \

.config("spark.hadoop.fs.s3a.aws.credentials.provider", "com.amazonaws.auth.DefaultAWSCredentialsProviderChain") \

.getOrCreate()

```

Here we add the hadoop-aws package and specify the S3A filesystem. Spark’s default AWS credentials provider chain (environment, config files, or IAM role) will be used. You may also set spark.hadoop.fs.s3a.access.key and .secret.key directly if needed, but using roles or AWS_* variables is recommended.

Now you can read files on S3 just like local files. Spark supports many formats:

```

hide

# Read a CSV file with header and schema inference

csv_df = spark.read \

.option("header", True) \

.option("inferSchema", True) \

.csv("s3a://my-bucket/data/myfile.csv")

# Read a JSON file

json_df = spark.read.json("s3a://my-bucket/data/mydata.json")

# Read a Parquet file (with built-in schema)

parquet_df = spark.read.parquet("s3a://my-bucket/data/mydata.parquet")

csv_df.show(5)

```

Spark will transparently split and parallelize reading from S3 (each worker can fetch data independently). Use wildcards or folders to read multiple files (e.g. "s3a://my-bucket/data/*.csv"). If your dataset is large, consider partitioning it (by date, ID, etc.) so Spark can skip irrelevant data. For example, if files are stored under …/year=2024/month=06/, Spark will only list the needed directories when you filter by year/month. Caching frequently-used tables with df.cache() can also speed up repeated reads.

Credentials and Configuration: You can supply AWS credentials in several ways: environment variables, IAM roles (when running on EC2/EMR), or the AWS CLI credentials file. Ensure you set spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem (Spark 3.x does this by default). Always match the hadoop-aws package version to your Hadoop version. For example, EMR 6.x uses Hadoop 3.x, so a 3.3.1 package is appropriate.

Writing to S3

Writing DataFrames to S3 is similar. Use the DataFrameWriter .write API with an S3 path. For example:

```

hide

# Overwrite or append CSV, JSON, Parquet

df.write.mode("overwrite").option("header", True) \

.csv("s3a://my-bucket/output/data.csv")

df.write.mode("overwrite") \

.json("s3a://my-bucket/output/data.json")

df.write.mode("overwrite") \

.parquet("s3a://my-bucket/output/data.parquet")

```

You can also specify compression or partitioning. For example, to write Parquet with Snappy compression and partition by year and month:

```

hide

df.write.mode("overwrite") \

.option("compression", "snappy") \

.partitionBy("year","month") \

.parquet("s3a://my-bucket/output/partitioned-data")

```

This creates subfolders like …/year=2024/month=06/. Partitioning improves query performance by enabling partition pruning. Similarly, you can use .format("csv") or .format("json") for more options. By default Parquet uses Snappy; you can change it to GZIP (.option("compression","gzip")) or other codecs.

When writing many files, be mindful of small-file issues. Coalesce or repartition your DataFrame to control the number of output files. For example, df.coalesce(10).write.parquet(...) ensures only 10 part files. Also use the EMR S3-optimized committer (enabled by default on EMR 5.19+ and Glue 3.0) to speed up writes: it uses multipart uploads instead of slow rename operations. If using AWS Glue, you can invoke a Glue-specific sink like:

```

hide

glueContext.write_dynamic_frame.from_options(

frame=dynamic_frame,

connection_type="s3",

format="glueparquet",

connection_options={"path": "s3://my-bucket/output", "partitionKeys": ["type"]},

format_options={"compression": "gzip"}

)

```

which similarly partitions and compresses data. (This uses AWS Glue’s DynamicFrame API.)

Using AWS Glue with Spark

AWS Glue is a managed ETL service built on Spark. You can write PySpark scripts as Glue jobs. In a Glue script, you import AWS Glue libraries and create a GlueContext, which wraps Spark. For example:

```

hide

from awsglue.utils import getResolvedOptions

from awsglue.context import GlueContext

from awsglue.job import Job

from pyspark.context import SparkContext

args = getResolvedOptions(sys.argv, ["JOB_NAME"])

sc = SparkContext()

glueContext = GlueContext(sc) # Glue-aware Spark context

spark = glueContext.spark_session

job = Job(glueContext)

job.init(args["JOB_NAME"], args)

```

Example AWS Glue Studio script editor: imports and initialization of GlueContext and Job. In Glue jobs, you typically start by initializing a GlueContext, which provides AWS-specific ETL functions, along with the normal SparkContext.

Once set up, you can use Glue’s DynamicFrame API to extract, transform, and load data. For example, to read from a Glue Data Catalog table (configured via crawlers or previous jobs):

```

hide

# Read from Glue Catalog (DynamicFrame)

dyf = glueContext.create_dynamic_frame.from_catalog(

database="my_database",

table_name="source_table",

transformation_ctx="dyf_source"

)

# (Optionally) convert to Spark DataFrame for complex transformations

df = dyf.toDF()

filtered_df = df.filter(df["value"] > 0)

# Convert back to DynamicFrame if needed

dyf2 = DynamicFrame.fromDF(filtered_df, glueContext, "filtered")

# Write back to S3 via Glue (partitioned and compressed Parquet)

glueContext.write_dynamic_frame.from_options(

frame=dyf2,

connection_type="s3",

format="parquet",

connection_options={"path": "s3://my-bucket/glue-output/", "partitionKeys": ["category"]},

format_options={"compression": "gzip"}

)

job.commit()

```

This script reads data from a Glue Catalog table, filters it, and writes out partitioned Parquet back to S3. AWS Glue provides many built-in transforms (like ApplyMapping, Join, etc.) and job bookmark features for incremental loads. By default Glue’s DynamicFrames integrate with the Glue Data Catalog. You can also use spark.read or spark.sql inside a Glue job once you have a SparkSession (via glueContext.spark_session).

Glue Job Setup: In the AWS Console, create a new Glue job (choose Spark/Python). You can edit the script in Glue Studio or upload one. The script must initialize GlueContext and Job as shown above. When finished, Glue will run it on a managed Spark cluster. You need an IAM role for Glue to access S3 and Glue Catalog. If you enable Glue Data Catalog integration (the default), your Glue job can access tables in the catalog without extra config.

Glue Catalog + Spark SQL

The AWS Glue Data Catalog can serve as a persistent Hive metastore for Spark. To use it, configure your Spark session or Glue job to point to the Glue Catalog. In AWS Glue or Glue Development Endpoints, you can pass the argument --enable-glue-datacatalog to enable this feature. On EMR, you set Spark’s Hive metastore class to the AWS Glue client factory:

{

"Classification": "spark-hive-site",

"Properties": {

"hive.metastore.client.factory.class":

"com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"

}

}

. This makes SparkSQL use Glue under the hood. Once enabled, you can run Spark SQL on tables in your Glue Catalog: for example:

```

hide

spark.sql("USE my_database").show()

spark.sql("SHOW TABLES").show()

spark.sql("SELECT * FROM customers LIMIT 5").show()

```

. These queries will list and read Glue tables as if they were Hive tables. Note that you may need to add the correct SerDe JARs if your tables use uncommon formats (Glue provides common ones for JSON, XML, etc.).

You can also bridge Glue DynamicFrames to SQL. For example, load a Glue table as a DynamicFrame and then run SQL on it:

```

hide

dyf = glueContext.create_dynamic_frame.from_catalog(database="mydb", table_name="mytable")

dyf.toDF().createOrReplaceTempView("mytable_view")

spark.sql("SELECT col1, count(*) FROM mytable_view GROUP BY col1").show()

```

. This leverages Glue-managed metadata in SparkSQL. In short, with the Glue Catalog enabled as Hive metastore, Spark SQL and Glue ETL scripts can both operate on the same tables seamlessly.

EMR with PySpark

Amazon EMR is a managed Hadoop/Spark service for running PySpark at scale. To run PySpark on EMR:

  1. Create an EMR cluster with Spark installed. In the EMR console, click “Create cluster,” select an EMR release that includes Spark (EMR 6.x or 7.x), and choose Spark in the applications. Configure instance types and count as needed. You can also enable “Use AWS Glue Data Catalog for Spark table metadata” in the console or via config (see above) so that Spark on EMR can use the Glue Catalog. Specify your EC2 key pair and IAM roles. Optionally enable S3 logging.
  2. Upload your PySpark script to S3 (or copy it onto the master node). It’s common to store scripts in an S3 bucket accessible to the cluster.
  3. Submit the job. You can add a step in the EMR console: go to your running cluster, open the Steps tab, and click Add step. Choose “Spark application” as the step type. For the application location, give the S3 path to your script, and any arguments. This runs spark-submit on your behalf. Select “client” or “cluster” deploy mode (cluster mode is typical). EMR will show the step’s progress and logs.

Alternatively, use the AWS CLI to submit steps. For example:

```

hide

aws emr add-steps --cluster-id <j-XXXX> \

--steps Type=Spark,Name="PySpark Step",ActionOnFailure=CONTINUE,\

```

Args=[--deploy-mode,cluster,--master,yarn,s3://my-bucket/scripts/job.py,arg1,arg2]

This tells EMR to run `spark-submit` with your script on the cluster.

4. **View results**: On completion, EMR will write logs to the configured S3 bucket (if logging enabled). You can also SSH to the master node and look at `/var/log/spark/` for driver stdout/stderr. The Spark UI (accessible via the EMR console) shows executors, stages, etc.

An example `spark-submit` command on EMR (inside a script or step) might look like:

```

hide

bash

spark-submit \

--deploy-mode cluster \

--master yarn \

s3://my-bucket/scripts/my_job.py \

--arg1 value1 --arg2 value2

```

This runs your PySpark application in YARN cluster mode. Make sure your EMR version has a compatible Python version (EMR 6.x uses Python 3 by default).

S3 Performance Tips

When using S3 with Spark/Glue, keep in mind that S3 is an object store (not a filesystem) and has its own performance characteristics. Here are some best practices:

More Spark tutorials

All tutorials · Try the free PySpark compiler · Practice challenges