Welcome to spark-on-aws-lambda v0.4.0 Discussions!

[JohnChe88]

👋 Welcome!

We’re using Discussions as a place to connect with other members of our community. We hope that you:

• Ask questions you’re wondering about.
• Share ideas.
• Engage with other community members.
• Welcome others and are open-minded. Remember that this is a community we
  build together 💪.

To get started, comment below with an introduction of yourself and tell us about what you do with this community.

[JohnChe88]

Our team is working on the version 0.2.0 currently in the develop branch. There were few bug in the v0.1.0. We are planning to release 0.2.0 end of the month. We are currently running stress test, regression test and trying different architectural patterns.

[guischuwarten]

Hi John,

How are you?

I would like ask you, if possible we execute spark on lambda with glue interation?

[JohnChe88]

@guischuwarten Thanks for asking. When you say Glue interaction, are you referring to Glue catalog? or AWS Glue? We have a working prototype for Glue catalog and we are testing it.

[JohnChe88]

Unfortunately the whole team is part of the AWS layoff, so please expect some delay in releasing new features

[andormarkus]

Hi @JohnChe88

Did the whole team lost the write access to this repo?
Can you still approve and merger PRs?

[JohnChe88]

@andormarkus Thanks for reaching out. I think my access looks good. Please create the PR and let me try

[paulorangeljr]

Hi there @JohnChe88

I've manage to shuffle data to s3 using

https://github.com/IBM/spark-s3-shuffle/packages/1834530

I think this is a good way to handle the limitations on lambda file read

Basically the spark submit I've added

"--conf", "spark.shuffle.manager=org.apache.spark.shuffle.sort.S3ShuffleManager", \
"--conf", "spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.S3ShuffleDataIO", \
"--conf", f"spark.shuffle.s3.rootDir=s3://{s3_bucket}/{s3_key}, \
"--conf", "spark.driver.extraClassPath=/var/lang/lib/python3.8/site-packages/pyspark/jars/spark-s3-shuffle_2.12-0.9.6-3.4.0.jar", \
"--conf", "spark.executor.extraClassPath=/var/lang/lib/python3.8/site-packages/pyspark/jars/spark-s3-shuffle_2.12-0.9.6-3.4.0.jar", \

[JohnChe88]

@paulorangeljr Interesting instead of spark submit we can also add it to the script. Is this application running in production?

[paulorangeljr]

@JohnChe88 we are making some final fine tuning in our dev environment but it is already mapped to be moved to production, the testing I believe it is done, we did an approach to use the "original" glue job whenever the lambda fails, so the pipeline is very very consistent

[JohnChe88]

@paulorangeljr Would you able to provide the company name so that we can add that to our list. This will help us train additional customer support agent to support SoAL. or please ask the AWS Account team to reach out to us?

[paulorangeljr]

@JohnChe88 I work for Johnson & Johnson Latam , but our Customer Support is from AWS US

[spongenee]

Hello! has anyone successfully read tables with millions of rows of data? What's your spark config? I keep getting the error doing select query on larger data sets

ERROR RetryingBlockTransferor: Exception while beginning fetch of 1 outstanding blocks
java.io.IOException: Failed to connect to /169.254.58.149:39573

[JohnChe88]

@spongenee

• what is the size of the file ? and how many total files
• Are you joining the table?
• What is Lambda memory that you are using ?

Based on the error you're encountering, this is a classic Spark shuffle networking issue that occurs when processing large datasets in AWS Lambda's constrained environment. Here's the specific solution:

Root Cause Analysis:

Executor failure during shuffle operations - Executors are dying due to memory pressure when processing millions of rows

Lambda's local-mode limitations - SoAL runs in single-node local mode, not distributed cluster mode

Insufficient memory allocation for large dataset operations

[spongenee]

what is the size of the file ? and how many total files

I am using a jdbc connection to a postgresql aurora cluster and reading one table, the query gets a designated amount of rows to a df and perform transformation and write to another PG aurora cluster. the code is in my forked repo https://github.com/spongenee/spark-on-aws-lambda/blob/jdbc-connection-pii-checker/spark-scripts/pg_column_pii_detector_incremental.py

Are you joining the table?

Yes, this is probably an intensive one: df.drop(column_name).join(spark.createDataFrame(anonymized_df, [PRIMARY_KEY, column_name]), PRIMARY_KEY,"leftouter").select(columns)

What is Lambda memory that you are using ?

10240MB

I just saw the above thread and will give spark-s3-shuffle a try and see if that helps!

[spongenee]

I have a specific question: how can we increase driver memory? I tried adding the flag to the spark submit command and got an error that is not quite informative

        subprocess.run([
            "spark-submit",
            "--driver-memory", "4g",
            "/tmp/spark_script.py",
            "--event", json.dumps(event)
        ], check=True, env=os.environ)

[ERROR] CalledProcessError: Command '['spark-submit', '--driver-memory', '4g', '/tmp/spark_script.py']' returned non-zero exit status 1.

[JohnChe88]

The --driver-memory flag issue you're encountering is due to Spark on AWS Lambda (SoAL) running in local mode, not cluster mode. In local mode, driver memory configuration works differently than traditional Spark clusters.

Solution for driver memory:

Instead of using --driver-memory in spark-submit, configure memory through these methods:

Set via SparkConf directly in your script:```python
from pyspark.sql import SparkSession
from pyspark.conf import SparkConf

conf = SparkConf()
conf.set("spark.driver.memory", "4g")
conf.set("spark.driver.maxResultSize", "2g") # Important for large datasets
conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

spark = SparkSession.builder.config(conf=conf).getOrCreate()

text

2. Increase Lambda function memory allocation:[3]
   • Lambda memory: 8-10GB (use maximum 10,240 MB)
   • This gives you ~6 vCPUs at max memory[3]
   • Higher memory = more CPU power for Spark processing

Regarding Large Datasets (Millions of Rows)

The IOException: Failed to connect error suggests network/shuffle issues typical with large datasets in Lambda's constrained environment.

Current limitations of SoAL:[2][1]

• Optimal for: <500MB payloads
• 15-minute timeout limit[4][5]
• Local mode only (single node processing)

Solutions for millions of rows:

1. Use S3 shuffle for large datasets:[6]

conf.set("spark.shuffle.manager", "org.apache.spark.shuffle.sort.S3ShuffleManager")
conf.set("spark.shuffle.sort.io.plugin.class", "org.apache.spark.shuffle.S3ShuffleDataIO")
conf.set("spark.shuffle.s3.rootDir", f"s3://{bucket}/{key}")
Optimize partitioning:

python
conf.set("spark.sql.shuffle.partitions", "50")  # Reduce from default 200
conf.set("spark.default.parallelism", "8")      # Match Lambda vCPUs
Memory optimization:

python
conf.set("spark.memory.fraction", "0.8")
conf.set("spark.memory.storageFraction", "0.3")
conf.set("spark.executor.memory", "6g")  # Most of Lambda memory

[spongenee]

thank you for your response @JohnChe88, I appreciate it!!

But setting driver memory via SparkConf directly in my script didn't work. I used the example spark scripts to test, and although they set driver and executor memory to 5g, the memory used for the lambda function is always around 2G after configuring the lambda function memory to the maximum 10G. Memory Size: 10240 MB Max Memory Used: 2236 MB

spark = SparkSession.builder.appName("SourceDBSession") \
    .master("local[*]") \
    .config("spark.driver.bindAddress", "127.0.0.1") \
    .config("spark.driver.memory", "5g") \
    .config("spark.executor.memory", "5g") \
    .getOrCreate()

And it seems to be expected according to pyspark's documentation, where it notes the below under spark.driver.memory. I believe setting master to local is indeed client mode.

Note: In client mode, this config must not be set through the SparkConf directly in your application, because the driver JVM has already started at that point. Instead, please set this through the --driver-memory command line option or in your default properties file.

[spongenee]

INFO ResourceUtils: No custom resources configured for spark.driver.
INFO Executor: Starting executor ID driver on host 169.254.9.65
INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 169.254.9.65, 33257, None)
INFO BlockManagerMasterEndpoint: Registering block manager 169.254.9.65:33257 with 289.6 MiB RAM, BlockManagerId(driver, 169.254.9.65, 33257, None)
INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 169.254.9.65, 33257, None)
INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 169.254.9.65, 33257, None)

Correct me if I am wrong. I hope the above lambda logs is not indicating that the driver RAM is only 289MB?

My spark conf, I have followed what you shared and included a few other custom ones

from pyspark.conf import SparkConf

conf = SparkConf()
conf.setMaster("local[*]")
conf.set("spark.driver.bindAddress", "127.0.0.1")
conf.set("spark.driver.memory", "5g")
conf.set("spark.executor.memory", "5g")
conf.set("spark.driver.maxResultSize", "2g")
conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
conf.set("spark.memory.fraction", "0.4")
conf.set("spark.network.maxRemoteBlockSizeFetchToMem", "1g")
conf.set("spark.task.maxDirectResultSize", "1g")
conf.set('spark.worker.cleanup.enabled', 'True')
conf.set('spark.sql.shuffle.partitions', 125)

[JohnChe88]

the “289MB” in your logs represents the Max Memory Used during the invocation(s), not that your Lambda only has 289MB allocated; the actual configured Memory Size is shown separately as “Memory Size: X MB.” If Memory Size is, say, 1024MB and Max Memory Used is 289MB, then the function peaked at ~289MB of RAM within that environment, leaving headroom that might be reducible depending on performance targets.

Actual physical memory might exactly translate teo spark in-memory
Start with a higher Lambda RAM to start with then once its starts working follow this,

Changes(after it works)

Adjust based on actual Lambda Memory Size; here’s a safer baseline for a function configured around 2–4GB:

conf.setMaster("local[*]") // keep
conf.set("spark.driver.bindAddress", "127.0.0.1") // keep
conf.set("spark.driver.memory", "1.5g") // tune to ~40–60% of Lambda Memory Size
Remove or shrink executor memory in local mode:
Either remove spark.executor.memory entirely, or set to a small value (e.g., "512m") if required by code paths.

conf.set("spark.driver.maxResultSize", "512m") // lower from 2g to avoid OOM on collect()
conf.set("spark.sql.execution.arrow.pyspark.enabled", "true") // keep if using Pandas UDFs; monitor memory

conf.set("spark.memory.fraction", "0.4") // fine; leave as-is
conf.set("spark.network.maxRemoteBlockSizeFetchToMem", "128m") // lower from 1g
conf.set("spark.task.maxDirectResultSize", "128m") // lower from 1g
conf.set("spark.worker.cleanup.enabled", "true") // keep
conf.set("spark.sql.shuffle.partitions", "48") // reduce from 125; tune based on data size

If Lambda Memory Size is higher (e.g., 6–10GB), scale spark.driver.memory proportionally but still leave ample headroom for non-heap memory.

[spongenee]

I've been setting Lambda memory to the maximum 10240MB, and then pushing the function to transform more lines at a time, the most I could do is 1.5 million rows. Once I up it to 1.6 million rows, I hit the error java.lang.OutOfMemoryError: Java heap space. When I get this error, however, in the Lambda log reporting, the maximum memory used is still ~2200MB.

I have tried setting spark.driver.memory to various values 1g, 2g, 3g, 5g, 6g but I keep seeing this in the reporting Max Memory Used: 2236 MB. That's why I was thinking maybe conf.set("spark.driver.memory", "5g") is not taking effect.