performance

结果表明，在Databricks中，Parquet（默认格式）的数据处理速度比ORC快

Optimize

• Prewarm clusters
  • by using Databricks Pool
  • target: reduce cluster start and auto-scaling times
• Prewarm caches
  • by running specific queries to initialize caches
• Specifying schema on flat file(e.g. csv) can improve dataframe read query performance
• Define how to join
  • by df.hint({broadcast/merge}) [TODO]
  • SortMerge: used to large data volume (larger than spark.sql.autoBroadcastJoinThreshold)
  • BroadcastHash: used for small data volume.
    • reason: smaller tables are transferred to every executor on the worker nodes, and cached locally in memory, and thus avoid data shuffle between the nodes
• Partition: trade-off between parallelism and data shuffles
  • Default partition block size is 128 MB
    • configure by spark.conf.set("spark.sql.files.maxPartitionBytes", 1048576) (for 1MB here)
  • Partition within a dataframe
    • When the data is small, then the number of partitions should be reduced
    • view created partitions count by df.rdd.getNumPartitions()
    • update by df.repartition(target_partitions_count).
      • Expensive operation: It involves fully shuffling the data
    • update by df.coalesce(minimum_partitions_count)
      • cheaper: It only moves data from redundant nodes
  • global setting for data shuffling trigger by transform (e.g. join(), union())
    • spark.conf.set("spark.sql.shuffle.partitions", target_partitions)

Reduce only df.coalesce({number of partitions}), which usually runs faster because executor leaves data in a minimum number of partitions and only moves data from redundant nodes Control the number of output files by df.write.option("maxRecordsPerFile", {number of files e.g. 1000}) Sets the the number of partitions to use during data shuffling, which is triggered by data transformations such as join(), union(), groupByKey(), reduceBykey(), and so on. spark.conf.set("spark.sql.shuffle.partitions", {number of partitions})

Benchmarks

2017 Benchmark on aws

• TPC-DS v2.4 (1000 scale factor)
• aws machine
  • 44 vcpu, 336 GB memory
  • 11 r3.xlarge (1 driver, 10 workers)
• Databricks Runtime 3.0