Hive 和 Spark 分区策略剖析

load().map(…).filter(…).save()

load().map(…).filter(…).coalesce(10).save()

load().coalesce(10).map(…).filter(…).save()

val df = load().map(…).filter(…).cache()df.count()df.coalesce(10)

df.withColumn("rand", rand() % filesPerPartitionKey).repartition(100, $"key", $"rand")

df.withColumn("rand", rand() % 5).repartition(5*365, $"date", $"rand")

class HashPartitioner(partitions: Int) extends Partitioner {    def getPartition(key: Any): Int = key match {        case null => 0        case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)    }}

import java.time.LocalDate def hashCodeTuple(one: String, two: Int, mod: Int): Int = { val rawMod = (one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)}def hashCodeSeq(one: String, two: Int, mod: Int): Int = { val rawMod = Seq(one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def iteration(numberDS: Int, filesPerPartition: Int): (Double, Double, Double) = {  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, filesPerPartition*numberDS))  )   hashedRandKeys.size // Number of unique keys, with the random factor   val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq   groupedHashedKeys.size // number of actual sPartitions used   val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse     val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse   sortedKeyCollisions.size // number of sPartitions with a hashing collision   // (collisions, occurences)  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse     (    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,  sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble  )} val results = Seq(  iteration(365, 1),  iteration(365, 5),  iteration(365, 10),  iteration(365, 100),  iteration(365 * 2, 100),  iteration(365 * 5, 100),  iteration(365 * 10, 100)) val avgEfficiency = results.map(_._1).sum / results.lengthval avgCollisionRate = results.map(_._2).sum / results.lengthval avgSevereCollisionRate = results.map(_._3).sum / results.length (avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2%, 42%, 12.6%

df.withColumn("rand", rand() % 5).repartition(5*365*SCALING_FACTOR, $"date", $"rand")

import java.time.LocalDate def hashCodeTuple(one: String, two: Int, mod: Int): Int = { val rawMod = (one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def hashCodeSeq(one: String, two: Int, mod: Int): Int = { val rawMod = Seq(one, two).hashCode % mod rawMod + (if (rawMod < 0) mod else 0)} def iteration(numberDS: Int, filesPerPartition: Int, partitionFactor: Int = 1): (Double, Double, Double, Double) = {  val partitionCount = filesPerPartition*numberDS * partitionFactor  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, partitionCount))  )     hashedRandKeys.size // Number of unique keys, with the random factor   val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq   groupedHashedKeys.size // number of unique hashes - and thus, sPartitions with > 0 records     val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse     val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse   sortedKeyCollisions.size // number of sPartitions with a hashing collision   // (collisions, occurences)  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse     (    groupedHashedKeys.size.toDouble / partitionCount,    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,    sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble  )} // With a scale factor of 1val results = Seq(  iteration(365, 1),  iteration(365, 5),  iteration(365, 10),  iteration(365, 100),  iteration(365 * 2, 100),  iteration(365 * 5, 100),  iteration(365 * 10, 100)) val avgEfficiency = results.map(_._2).sum / results.length // What is the ratio of executors / output filesval avgCollisionRate = results.map(_._3).sum / results.length // What is the average collision rateval avgSevereCollisionRate = results.map(_._4).sum / results.length // What is the average collision rate where 3 or more hashes collide (avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2% Efficiency, 42% collision rate, 12.6% severe collision rate iteration(365, 5, 2) // 37.7% partitions in-use, 77.4% Efficiency, 24.4% collision rate, 4.2% severe collision rateiteration(365, 5, 5)iteration(365, 5, 10)iteration(365, 5, 100)