orderby在Spark SQL中未给出正确的结果

如何解决orderby在Spark SQL中未给出正确的结果

我有大约60列和3000行的数据集。我正在使用orderby对数据集中的行进行排序并写入文件但是它并没有给出正确的结果。

dataset.orderBy(new Column(col_name).desc())
                .coalesce(4)
                .write()
                .format("com.databricks.spark.csv")
                .option("delimiter",",")
                .option("header","false")
                .mode(SaveMode.Overwrite)
                .save("hdfs://" + filePath);

请让我知道我在这里想念的东西

我也找到了下面的解决方案，但不认为这是正确的解决方案

        Row[] rows = dataset.take(3000);

        for ( Row row : rows){
            // here i am writing in a file row by row
            System.out.println(row);
        }

解决方法

问题在于coalesce将以未排序的方式合并您现有的分区（不会，coalesce不会造成混乱）。

如果要4个文件并在文件内排序，则需要在spark.sql.suffle.partitions之前更改orderBy，这将导致随机播放具有4个分区。

spark.sql("set spark.sql.shuffle.partitions=4")

dataset.orderBy(new Column(col_name).desc())
            .write()
            .format("com.databricks.spark.csv")
            .option("delimiter",",")
            .option("header","false")
            .mode(SaveMode.Overwrite)
            .save("hdfs://" + filePath);

如果您只关心文件内的排序，则也可以使用sortWithinPartitions(new Column(col_name).desc())

因为您的.coalesce(4)拖延了数据框顺序

先行然后排序。

dataset
.coalesce(4)
.orderBy(new Column(col_name).desc())                
.write()
.format("com.databricks.spark.csv")
.option("delimiter",")
.option("header","false")
.mode(SaveMode.Overwrite)
.save("hdfs://" + filePath);

在火花上下文中，您还应该将spark.sql.suffle.partitions设置为4，因为按顺序还可以使用先驱顺序。

根据您在注释中的澄清，您需要将ordered输出包含在一个文件中。

只有火花，只有spark.sql("set spark.sql.shuffle.partitions=1")后跟orderBy并写入才有可能。但是缺点是won't scale不能用于大数据，因为它不会并行化。

一种解决方法是：

通过最大程度的并行化工作（例如，不要orderBy或coalesce）来"set spark.sql.shuffle.partitions=1"，并拥有n个文件。
在文件合并代码中添加一些额外的逻辑处理
列出所有文件，获取col_name的值并维护[(col_name value),filepath]的映射
通过键（col_name的值）对地图进行排序
然后执行合并

这将保持您的订购。

想法是，合并部分将大部分为单线程，至少以分布式方式进行排序：）