Spark using python: How to resolve Stage x contains a task of very large size (xxx KB). The maximum recommended task size is 100 KB

by

The general idea is that PySpark creates as many java processes than there are executors, and then ships data to each process. If there are too few processes, a memory bottleneck happens on the java heap space.

In your case, the specific error is that the RDD that you created with sc.parallelize([...]) did not specify the number of partition (argument numSlices, see the docs). And the RDD defaults to a number of partition that is too small (possibly it is constituted by a single partition).

To solve this problem, simply specify the number of partitions wanted:

a = sc.parallelize([...], numSlices=1000)   # and likewise for b

As you specify higher and higher number of slices, you will see a decrease in the size stated in the warning message. Increase the number of slices until you get no more warning message. For example, getting

Stage 0 contains a task of very large size (696 KB). The maximum recommended task size is 100 KB

means that you need to specify more slices.

Another tip that may be useful when dealing with memory issues (but this is unrelated to the warning message): by default, the memory available to each executor is 1 GB or so. You can specify larger amounts through the commandline, for example with --executor-memory 64G.

More Related Contents:

  1. Spark Dataframe validating column names for parquet writes
  2. How to save/insert each DStream into a permanent table
  3. reading json file in pyspark
  4. How to optimize shuffle spill in Apache Spark application
  5. Does a join of co-partitioned RDDs cause a shuffle in Apache Spark?
  6. Spark SQL: apply aggregate functions to a list of columns
  7. Unpivot in spark-sql/pyspark
  8. How to stop INFO messages displaying on spark console?
  9. DataFrame join optimization – Broadcast Hash Join
  10. Overwrite specific partitions in spark dataframe write method
  11. Convert date from String to Date format in Dataframes
  12. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  13. What is the difference between spark checkpoint and persist to a disk
  14. How to write spark streaming DF to Kafka topic
  15. Spark: Reading files using different delimiter than new line
  16. How to fix ‘TypeError: an integer is required (got type bytes)’ error when trying to run pyspark after installing spark 2.4.4
  17. Spark MLlib LDA, how to infer the topics distribution of a new unseen document?
  18. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  19. Multiple Spark applications with HiveContext
  20. How to loop through each row of dataFrame in pyspark
  21. How to group by common element in array?
  22. Understanding Spark serialization
  23. Pyspark: Pass multiple columns in UDF
  24. Convert null values to empty array in Spark DataFrame
  25. DataFrame partitionBy to a single Parquet file (per partition)
  26. Temp table caching with spark-sql
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. How to calculate Median in spark sqlContext for column of data type double
  29. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’
  30. Understanding Spark’s caching

Leave a Comment