Spark 2.1中用partitionBy()和persist()优化脚本遇未知错误求助

Hey there! Let’s dive into common pitfalls and fixes when optimizing your Spark 2.1 script with hash partitioning and persist(). I’ll cover typical mistakes, correct implementation patterns, and how to resolve frequent errors you might run into.

Common Mistake 1: Misusing Hash Partitioning #

Hash partitioning shines when you target columns that are frequently used in joins, filters, or aggregations. A common misstep is either picking the wrong column or using the incorrect Spark method.

Correct Implementation Example #

For DataFrames:

// Repartition by a high-cardinality key column using hash partitioning
val partitionedDF = originalDF.repartition(12, $"user_id") // 12 partitions, hash on user_id

For RDDs:

val partitionedRDD = originalRDD.partitionBy(new HashPartitioner(12))

Typical Error: Using partitionBy on DataFrames (which is for bucketing, not hash partitioning) instead of repartition. This triggers errors like org.apache.spark.sql.AnalysisException: 'partitionBy' is not a valid method for DataFrame.

Common Mistake 2: Wrong Order of Operations (Persist vs Partition) #

Persisting should always happen after you apply partitioning—otherwise you’re caching unoptimized data and wasting resources. Choosing the wrong storage level also leads to avoidable issues.

Correct Order Example #

// First partition, then persist with a suitable storage level
val optimizedDF = originalDF.repartition(12, $"user_id")
  .persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK) // Balances memory/disk usage

// Reuse the optimized DataFrame across multiple operations
val userAggs = optimizedDF.groupBy($"user_id").agg(sum($"purchase_amount").as("total_spent"))
val activeUsers = optimizedDF.filter($"last_login" > "2024-01-01")

// Clean up resources when done
optimizedDF.unpersist()

Typical Error: Persisting before partitioning forces Spark to recompute partitions later, negating your optimization. Using MEMORY_ONLY for large datasets causes OutOfMemoryError—switch to MEMORY_AND_DISK to spill excess data to disk.

Common Mistake 3: Serialization Failures with Persist() #

Spark 2.1 defaults to Java serialization, which can fail if your data types aren’t serializable or are slow to process.

Fix: Switch to Kryo Serialization #

Add these configs to your SparkSession:

val spark = SparkSession.builder()
  .appName("OptimizedSparkApp")
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.kryo.registrationRequired", "true") // Optional but enforces type safety
  .getOrCreate()

Typical Error: java.io.NotSerializableException when calling persist(). Register custom classes with Kryo if you’re working with non-standard data types.

Spark 2.1 Specific Quirks #

• Complex Type Partitioning: Spark 2.1 has limited support for hash partitioning on complex types (structs, arrays). Partitioning on these columns leads to uneven data distribution or performance hits—stick to primitive columns (strings, integers) instead.
• Bucketing + Persist: If you’re using bucketing via bucketBy, avoid calling persist() directly on the bucketed DataFrame. Spark handles caching for bucketed tables automatically in most cases.

Example Error Resolution #

Suppose you see this error:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 5, node-1): java.lang.OutOfMemoryError: Java heap space

This is almost always due to using MEMORY_ONLY for a large partitioned dataset. Fix it by updating your storage level:

partitionedDF.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK)

Alternatively, adjust partition count (if partitions are too small) or increase executor memory in your Spark cluster config.

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