Spark 数据框/数据集上的自定义排序顺序答案

【问题标题】：Custom sort order on a Spark dataframe/datasetSpark 数据框/数据集上的自定义排序顺序
【发布时间】：2020-01-28 05:04:17
【问题描述】：

我有一个围绕 Spark 构建的 Web 服务，它基于 JSON 请求构建一系列数据帧/数据集操作。

这些操作涉及多个连接、过滤器等，它们会改变列中值的顺序。这个最终的数据集可能有数百万行。

最好不将其转换为 RDD，是否可以根据作为列表传入的元素的顺序对最终数据集的某些列应用自定义排序？

原始数据框的格式为

+----------+----------+
| Column 1 | Column 2 |
+----------+----------+
| Val 1    | val a    |
+----------+----------+
| Val 2    | val b    |
+----------+----------+
| val 3    | val c    |
+----------+----------+

执行一系列转换后，数据框最终看起来像这样。

+----------+----------+----------+----------+
| Column 1 | Column 2 | Column 3 | Column 4 |
+----------+----------+----------+----------+
| Val 2    | val b    | val 999  | val 900  |
+----------+----------+----------+----------+
| Val 1    | val c    | val 100  | val 9$#@ |
+----------+----------+----------+----------+
| val 3    | val a    | val 2##  | val $#@8 |
+----------+----------+----------+----------+

我现在需要根据作为数组列表传递的值的顺序对多列应用排序。

例如：
Col1values 顺序=[val 1,val 3,val 2}
Col3values Order=[100,2##,999]。

【问题讨论】：

我想通过自定义排序顺序，您希望将某种函数传递给sort()。但是，计算该排序键并将其添加为列进行排序是否可能，甚至更容易？
请提供一些您到目前为止所做的代码，并提供 IP 数据和预期 OP 的示例。因为似乎不清楚你想要达到什么目的。
我觉得这个帖子会对你有所帮助。 stackoverflow.com/questions/55128213/…
您能否提供有关“自定义排序”的更多详细信息？
我已编辑问题以添加更多详细信息。

标签： java apache-spark collections java-8 apache-spark-sql

【解决方案1】：

自定义排序通过创建用于排序的列来工作。它不需要是数据框中的可见列。我可以使用 PySpark 展示它。

初始df：

from pyspark.sql import functions as F
df = spark.createDataFrame(
    [(1, 'a', 'A'),
     (2, 'a', 'B'),
     (3, 'a', 'C'),
     (4, 'b', 'A'),
     (5, 'b', 'B'),
     (6, 'b', 'C'),
     (7, 'c', 'A'),
     (8, 'c', 'B'),
     (9, 'c', 'C')],
    ['id', 'c1', 'c2']
)

在 1 列上自定义排序：

from itertools import chain

order = {'b': 1, 'a': 2, 'c': 3}

sort_col = F.create_map([F.lit(x) for x in chain(*order.items())])[F.col('c1')]
df = df.sort(sort_col)

df.show()
# +---+---+---+
# | id| c1| c2|
# +---+---+---+
# |  5|  b|  B|
# |  6|  b|  C|
# |  4|  b|  A|
# |  1|  a|  A|
# |  2|  a|  B|
# |  3|  a|  C|
# |  7|  c|  A|
# |  8|  c|  B|
# |  9|  c|  C|
# +---+---+---+

在 2 列上：

from itertools import chain

order1 = {'b': 1, 'a': 2, 'c': 3}
order2 = {'B': 1, 'C': 2, 'A': 3}

sort_col1 = F.create_map([F.lit(x) for x in chain(*order1.items())])[F.col('c1')]
sort_col2 = F.create_map([F.lit(x) for x in chain(*order2.items())])[F.col('c2')]
df = df.sort(sort_col1, sort_col2)

df.show()
# +---+---+---+
# | id| c1| c2|
# +---+---+---+
# |  5|  b|  B|
# |  6|  b|  C|
# |  4|  b|  A|
# |  2|  a|  B|
# |  3|  a|  C|
# |  1|  a|  A|
# |  8|  c|  B|
# |  9|  c|  C|
# |  7|  c|  A|
# +---+---+---+

或者作为一个函数：

from itertools import chain
def cust_sort(col: str, order: dict):
    return F.create_map([F.lit(x) for x in chain(*order.items())])[F.col(col)]

df = df.sort(
    cust_sort('c1', {'b': 1, 'a': 2, 'c': 3}),
    cust_sort('c2', {'B': 1, 'C': 2, 'A': 3})
)

【讨论】：