1

I observe severe underutilization of CPU in my Databricks job run metrics, on average less than 50% - indicating that I do not parallelize enough tasks in the Spark workflow.

I am especially interested in improving the job's read parallelism. For context, I read multiple tables.

Is my knowledge correct that, concerning the read stage, Spark creates the same amount of tasks as the table-to-read has partitions? Moreover, what does this look like when my table is not partitioned?

I know that by default the configuration option spark.sql.shuffle.partitions = 200. However, the Spark docs detail that this property applies to wide transformations (i.e. joins & aggregates) happening in shuffle stages which come after the initial read stage if I am not mistaken.

Configures the number of partitions to use when shuffling data for joins or aggregations.

Since reading is neither a join nor an aggregation, I wonder if this default value (i.e. 200) will be the level of Spark parallelism (i.e. 200 tasks) to fill up the 6*16 = 96 cores in my cluster to read that unpartitioned data.

On the unpartitioned table, I have messed with spark.conf.set("spark.sql.files.maxPartitionBytes", "64MB") that controls the maximum number of bytes read per partition (default: 128 MB) to increase the number of Spark partitions, which increases the number of tasks and, therefore, parallelism.

However, I have not seen a significant result in performance.

Improve this question
2
  • When reading a partitioned table, Spark creates one task per partition. If a table has 100 partitions, Spark will create 100 tasks for reading. Commented Feb 24 at 15:11
  • @DileepRajNarayanThumula, thank you for the comment; this confirms my first question. How about unpartitioned tables? Commented Feb 24 at 15:21

1 Answer 1

Reset to default
1

Whether a (Delta) table is partitioned or not or reading a pruned partition, there is a common approach.

  1. spark.sql.shuffle.partitions has no role in reading from data at rest that is relevant for shuffling due JOIN etc.
  2. Does not matter if pyspark or Spark with Scala API.
  3. spark.default.parallelism is only relevant for RDD's so leaving out of the equation.
  4. spark.sql.files.maxPartitionBytes is relevant and also if the source is a parquet file one could argue.

Imagine a Delta table with 5 files: 1 very large file (X size) and 4 smaller files (X/5 size). The App has 2 vCores available, and spark.sql.files.maxPartitionBytes = X/2.

Then:

  1. The number of Tasks/parallelization will be:
    • 4 for the smaller files that are of size X/5 that fit into X/2 sizing
    • 2 for the large file of size X, that is split on X/2
  2. 6 Tasks can run concurrently to read the file.
Improve this answer
Sign up to request clarification or add additional context in comments.

1 Comment

Thank you for the answer. I suppose that the example you gave is for partitioned tables, where different partitions may vary greatly in size. To my understanding, the file sizes of an unpartitioned table are more or less uniform, and tweaking spark.sql.files.maxPartitionBytes will control the read parallelism in that case.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.