spark中的预分区数据，这样每个分区在我们正在分区的列中都有不重叠的值

Question

我试图在对数据的某一列执行聚合操作之前对数据进行预分区。我有3个工作节点，我希望每个分区在我分区的列中都有不重叠的值。我不希望出现两个分区在列中可能具有相同值的情况。

例如。如果我有以下数据

ss_item_sk | ss_quantity
1          | 10.0
1          |  4.0
2          |  3.0
3          |  5.0
4          |  8.0
5          |  13.0
5          |  10.0

那么以下隔墙是令人满意的：

分区1

ss_item_sk | ss_quantity
1          | 10.0
1          |  4.0

分区2

ss_item_sk | ss_quantity
2          |  3.0
3          |  5.0

分区3

ss_item_sk | ss_quantity
4          |  8.0
5          |  13.0
5          |  10.0

不幸的是，我下面的代码不起作用。

spark.sqlContext.setConf( "spark.sql.shuffle.partitions", "3")
var json = spark.read.json("hdfs://master:9000/tpcds/store_sales")
var filtered = json.filter(row => row.getAs[Long]("ss_item_sk") < 180)
filtered.repartition($"ss_item_sk").write.json(savepath)

我已经看过了

• 如何定义数据帧的分区

我还是想不通。

Answer 1

按键重新分区根据数据帧级别的键进行数据分布。而在hdfs上编写数据帧是一件独立的事情

df.coalesce(1).write.partitionBy("ss_item_sk").json(savepath)

在这个场景中，您也将看到分区列创建的不同目录中的多个零件文件。只能根据“partitionBy”方法控制将运行的写入程序/还原程序的数量。它与Map Reduce Partitioner非常相似，因为它控制着要运行的reducer的数量。要基于分区列获取单个文件，必须运行以下命令。

df.repartition($"ss_item_sk").write.partitionBy("ss_item_sk").json(savepath)

现在，当减速器被映射到执行器分区的数量时，这就起作用了。希望这能有所帮助