Hubert-Dudek
Databricks MVP

Be sure to import as "import pyspark.pandas as ps"

Please compare times with similar operations on the usual spark first. There can be multiple problems related to the dataset.

  • Look for data skews; some partitions can be huge, some small because of incorrect partitioning. You can use Spark UI to do that but also debug your code a bit (get getNumPartitions()) especially, SQL can divide it unequally to partitions (there are settings in connector lowerBound, etc.). You could try to have a number of partitions as workers cores multiplied by X (so they will be processed step by step in the queue). When data is processed, your partitions should be ft in RAM.
  • Increase shuffle size spark.sql.shuffle.partitions default is 200 try bigger. It would be best if you calculated it as data size divided by the size of the partition,
  • increase the size of the driver to be two times bigger than the executor (but to get the optimal size, please analyze load - in databricks on cluster tab look to Metrics there is Ganglia or even better integrate datadog with cluster),
  • Check wide transformations. Such transformations must shuffle data between partitions and group them to do only one shuffle. Suppose a shuffle happens (for example, getting the field's average value from the whole dataset will require data from all partitions). Remember that partitions are moving between workers using the network. There is already a partition in memory on workers, and there should be enough operation memory for incoming data. Writing to disk if there is no memory can be the bottleneck (check disk spills in Spark UI)
  • If you need to filter data, if possible, do it after reading from SQL so it will be predicative push and add where in SQL query.
  • Make sure that everything runs in a distributed way, specially UDF. It would help if you used vectorized pandas udfs so that they will run on executors. Don't use collect etc.
  • Sometimes (but I bet it is not that case), I process big data as a stream as it is easier with big data sets. In that scenario, you would need Kafka (which can be a confluent cloud) between SQL and Databricks.
  • Regarding infrastructure, use more workers and check that your ADLS is connecting through a private link. Monitor save progress in a folder. You can also use premium ADLS which is faster.


My blog: https://databrickster.medium.com/