由于spark对数据进行分区的方式，你将无法准确地实现这一点。Spark会接受你在repartition中指定的列，将该值散列为64 b长，然后将该值与分区数取模。这样，分区数是确定性的。它以这种方式工作的原因是，连接还需要连接左右两侧的分区数匹配以确保散列在两侧是相同的。
“我们希望为一个国家的每10个人创建一个分区。”
你到底想在这里完成什么？在一个分区中只有10行可能对性能很糟糕。你是否试图创建一个分区表，其中每个分区中的文件都保证只有x行？
“df.repartition（$“country”）：这将为中国创建一个分区，为法国创建一个分区，为古巴创建一个分区”
这实际上将创建一个具有默认的shuffle分区数的加密框架，该分区数按国家进行散列处理

def repartition(partitionExprs: Column*): Dataset[T] = {
    repartition(sparkSession.sessionState.conf.numShufflePartitions, partitionExprs: _*)
  }

“df.repartition（8，$“country”，兰德）：这将为每个国家创建最多8个分区，因此它应该为中国创建8个分区，但法国和古巴分区是未知的。法国可以在8个分区中，古巴可以在5个分区中。有关详细信息，请参阅此答案。”
像wise这是微妙的错误。只有8个分区与国家基本上随机 Shuffle 在这8个分区。
编辑：最后一点要澄清的是， Dataframe 重分区的工作方式与写入时不同，你做了一个partitionBy（...）方法。partitionBy操作spark首先获取所有的spark分区，而不是每个spark分区。它将其切片到一个表partitionBy partition中，然后将每个partitionBy分区写入与partitionBy列对应的文件夹。

下面的代码将为每个数据文件（sample dataset is here）创建10行：

val outputPath = new java.io.File("./tmp/partitioned_lake5/").getCanonicalPath
df
  .repartition(col("person_country"))
  .write
  .option("maxRecordsPerFile", 10)
  .partitionBy("person_country")
  .csv(outputPath)

以下是Spark 2.2之前的代码，它将为每个数据文件创建大约10行：

val desiredRowsPerPartition = 10

val joinedDF = df
  .join(countDF, Seq("person_country"))
  .withColumn(
    "my_secret_partition_key",
    (rand(10) * col("count") / desiredRowsPerPartition).cast(IntegerType)
  )

val outputPath = new java.io.File("./tmp/partitioned_lake6/").getCanonicalPath
joinedDF
  .repartition(col("person_country"), col("my_secret_partition_key"))
  .drop("count", "my_secret_partition_key")
  .write
  .partitionBy("person_country")
  .csv(outputPath)