Great question! Let's break down the optimizations you can apply, starting with your KryoSerializer question, then moving through read, transform, write, and resource tuning steps:

Is it effective? #

Yes, absolutely. The default Java serializer is slow and memory-inefficient for large datasets—KryoSerializer is typically 5-10x faster, uses less memory, and reduces garbage collection (GC) pressure. This is especially impactful for your large 92GB and 20GB tables, as it speeds up data serialization/shuffling across Glue nodes.

How to configure it #

You have two reliable ways to set this up for your Glue job:

1. Job Parameters (Recommended): When creating/editing your Glue job, add this to the Job parameters section:

   --conf spark.serializer=org.apache.spark.serializer.KryoSerializer
   

   If you’re using custom classes (unlikely in standard Glue workflows), you can register them with a Kryo registrator:

   --conf spark.kryo.registrator=com.yourorg.YourCustomRegistrator
   

2. Code-Based Configuration: Add this early in your script (before creating DataFrames):

   from pyspark.context import SparkContext
   sc = SparkContext.getOrCreate()
   sc._conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
   

How to validate it's working #

• Spark UI: After your job runs, open the Spark UI (from the Glue job details page) and navigate to the Storage tab. Check the "Serializer" column for cached RDDs/DataFrames—it should show KryoSerializer.
• CloudWatch Logs: Search your job's CloudWatch logs for the line Using KryoSerializer—this confirms Spark picked up the configuration.
• Performance Comparison: Run a test job with and without Kryo, then compare metrics like total runtime, GC time (look for logs with GC time elapsed), and memory usage. You should see a noticeable drop in GC overhead and faster execution.

Let’s break this down by stage to target bottlenecks:

Read Stage (Aurora Postgres) #

• Optimize JDBC Partitioning: Your numPartitions=100 is a good start, but verify your partitionColumn (date_col) has evenly distributed data. Run this query in Postgres to check:

  SELECT date_col, COUNT(*) 
  FROM first_largest_table 
  WHERE maturity_date > (current_date - 45)::timestamp without time zone AT TIME ZONE 'UTC'
  GROUP BY date_col
  ORDER BY COUNT(*) DESC;
  

  If some partitions have 10x more data than others, switch to a more evenly distributed column (like an auto-increment id column) to avoid skewed, slow tasks.
• Increase Fetch Size: Add .option("fetchSize", "10000") to your JDBC read configuration. The default fetch size is 1000, so increasing this reduces round-trips between Glue and Postgres.
• Pre-Join Small Tables in Postgres: Instead of joining multiple small tables in Glue, create a materialized view in Postgres that combines them (e.g., join the 68MB, 50MB, and smaller tables into one view). Reading a single view cuts down on Spark shuffle overhead.
• Verify Postgres Indexes: Ensure maturity_date and your partitionColumn have indexes in Postgres. This speeds up the initial filter and partitioning query, reducing the time Glue waits for data.

Transform Stage #

• Double-Check Broadcast Joins: You’re already using broadcasts, but confirm you’re applying them correctly:
  • Use from pyspark.sql.functions import broadcast and explicitly wrap small DataFrames: large_df.join(broadcast(small_df), on="id")
  • Adjust the auto-broadcast threshold if needed: Add --conf spark.sql.autoBroadcastJoinThreshold=70000000 (70MB) to your job parameters to cover your 68MB table.
• Detect and Fix Data Skew: If joins are slow, check for skewed join keys with:

  from pyspark.sql.functions import desc
  billing_fee_df.groupBy("id").count().orderBy(desc("count")).show(10)
  

  If a single id has millions of rows, use salting: Add a random suffix to the large table's join key, expand the small table's key to match all possible suffixes, join, then aggregate to remove the suffix.
• Cache Reused DataFrames: If you reuse any DataFrame multiple times (e.g., joining it with multiple tables), cache it with serialization to save memory:

  from pyspark.storagelevel import StorageLevel
  billing_fee_df.persist(StorageLevel.MEMORY_AND_DISK_SER)
  

• Early Column Pruning: Double-check that your JDBC query only selects columns you actually need—unnecessary columns increase data transfer and memory usage.

Write Stage (Redshift) #

• Use Glue's Redshift Format Instead of JDBC: The redshift format uses S3 as a staging layer (bulk loading) which is far faster than JDBC inserts. Here’s a sample configuration:

  billing_fee_df.write.format("redshift") \
    .option("url", "jdbc:redshift://your-cluster.us-west-2.redshift.amazonaws.com:5439/your-db") \
    .option("dbtable", "your_target_table") \
    .option("user", "redshift-user") \
    .option("password", "redshift-pass") \
    .option("tempdir", "s3://your-bucket/glue-temp/") \
    .option("compression", "snappy") \
    .mode("overwrite") \
    .save()
  

• Tune Write Parallelism: For your 2-5GB target table, set spark.sql.shuffle.partitions to 50-100 (add --conf spark.sql.shuffle.partitions=75 to job parameters). This ensures each partition is a reasonable size (20-100MB) for Redshift to load efficiently.
• Enable Compression: The snappy compression option reduces the size of staging files in S3, speeding up uploads and Redshift's COPY operation.

Resource & Cost Optimization #

• Right-Size Your Nodes: Check CloudWatch metrics for your job (CPUUtilization, MemoryUtilization):
  • If CPU is consistently <50%, reduce the number of nodes (e.g., from 10 to 7) or switch to smaller instances.
  • If memory is >80%, switch to memory-optimized instances (e.g., R.2X instead of G.1X) or enable KryoSerializer (which reduces memory footprint).
• Increase Concurrency (Carefully): Your current concurrency of 1-3 is low. Try increasing to 5-8, but first verify that your Aurora Postgres instance's max_connections can handle the load (each JDBC partition uses one connection—your numPartitions=100 means up to 100 concurrent connections).
• Use Glue Flexible Execution Mode: Flexible mode automatically scales nodes up/down based on workload, saving cost during idle stages (like waiting for Postgres data) and providing more resources during shuffle/write stages.
• Incremental Processing: If this is a recurring job, modify it to only process new/updated data (e.g., filter for maturity_date > last_job_run_date). This drastically reduces data volume and runtime.

Always check the Spark UI for your Glue job—it shows exactly where time is being spent (e.g., slow shuffle stages, skewed tasks, long read/write times). This is the best way to target your optimizations to the actual bottlenecks.

内容的提问来源于stack exchange，提问作者Virat Arya