如何根据价值在火花中映射多对多？

Question

对于输入数据框：

+-------+-----+-------+------+--------------+-------+----+
|Closing| Flow|Opening|Period|        RowKey|RowKey2|Year|
+-------+-----+-------+------+--------------+-------+----+
|  -2.11|-2.11|    0.0|    01|2016-01-1200-A| 1200-A|2016|
|  -1.11|-1.11|    0.0|    01|2016-01-1200-B| 1200-B|2016|
|   -1.0| -1.0|    0.0|    04|  2016-04-2200|   2200|2016|
|   -1.0| -1.0|    0.0|    04|  2016-04-3000|   3000|2016|
|   -1.0| -1.0|    0.0|    05|2016-05-1200-C| 1200-C|2016|
|    0.0|  1.0|   -1.0|    05|  2016-05-3000|   3000|2016|
|    0.0|  1.0|   -1.0|    08|  2016-08-2200|   2200|2016|
|    1.0|  1.0|    0.0|    09|  2016-09-2200|   2200|2016|
|   -2.0| -1.0|   -1.0|    12|2016-12-1200-C| 1200-C|2016|
|  100.0|100.0|    0.0|    12|  2016-12-4000|   4000|2016|
+-------+-----+-------+------+--------------+-------+----+

我想从下面应用函数（其中 period 是 input["Period] 列中不同周期值的列表）：

def insertPeriod(row, period):
    row = row.asDict()
    row["Period"]=period
    return Row(**row)

def filterOutFromPartion(aggdata, periods):
    output = []
    for period in periods:
        iterator = (item for item in aggdata if item["Period"] == period)
        try:
            found = next(iterator)
            output.append(insertPeriod(found,period))
        except StopIteration:
            if (len(output)==0):
                continue
            else:
                temp = output[-1]
                output.append(insertPeriod(temp,period))
    return iter(output)

所以结果是：

+--------------+----+------+-------------+-----+--------------+--------------+
|        RowKey|Year|Period|AccountNumber| Flow|OpeningBalance|ClosingBalance|
+--------------+----+------+-------------+-----+--------------+--------------+
|2016-01-1200-A|2016|     1|         1200|-2.11|             0|         -2.11|
|2016-01-1200-B|2016|     1|         1200|-1.11|             0|         -1.11|
|2016-02-1200-A|2016|     2|         1200|    0|         -2.11|         -2.11|
|2016-02-1200-B|2016|     2|         1200|    0|         -1.11|         -1.11|
|2016-03-1200-A|2016|     3|         1200|    0|         -2.11|         -2.11|
|2016-03-1200-B|2016|     3|         1200|    0|         -1.11|         -1.11|
|2016-04-1200-A|2016|     4|         1200|    0|         -2.11|         -2.11|
|2016-04-1200-B|2016|     4|         1200|    0|         -1.11|         -1.11|
| 2016-04-2200-|2016|     4|         2200|   -1|             0|            -1|
| 2016-04-3000-|2016|     4|         3000|   -1|             0|            -1|
|2016-05-1200-A|2016|     5|         1200|    0|         -2.11|         -2.11|
|2016-05-1200-B|2016|     5|         1200|    0|         -1.11|         -1.11|
|2016-05-1200-C|2016|     5|         1200|   -1|             0|            -1|
| 2016-05-2200-|2016|     5|         2200|    0|            -1|            -1|
| 2016-05-3000-|2016|     5|         3000|    1|            -1|             0|
|2016-06-1200-A|2016|     6|         1200|    0|         -2.11|         -2.11|
|2016-06-1200-B|2016|     6|         1200|    0|         -1.11|         -1.11|
|2016-06-1200-C|2016|     6|         1200|    0|            -1|            -1|
| 2016-06-2200-|2016|     6|         2200|    0|            -1|            -1|
| 2016-06-3000-|2016|     6|         3000|    0|             0|             0|
+--------------+----+------+-------------+-----+--------------+--------------+
only showing top 20 rows

基本上按 RowKey2 值对 rdd 进行映射操作，如果缺少关于 period 的信息，则只需使用最后一个的信息（如果存在）。

所以我很想使用

 df.rdd.partitionBy("RowKey2")\
.mapPartitions(lambda x: filterOutFromPartion(x, periodsList))\
.collect()

哪个上升

Py4JError：调用时出错 None.org.apache.spark.api.python.PythonPartitioner。痕迹： py4j.Py4JException：构造函数 org.apache.spark.api.python.PythonPartitioner([class java.lang.String, 类 java.lang.Long]) 不存在

如果我跳过按键分区，那么我只会得到 2 个前行键的结果（每个 12 个周期都符合预期）。有人可以在那里为我提供帮助吗？

问候， 迈克

Answer 1

partitionBy签名：

partitionBy(numPartitions, partitionFunc=)

其中第一个参数是整数，第二个（可选）是一个函数。没有接受字符串的变体。你可能把它和Dataframe.repartition混淆了

df.repartition(n, "RowKey2")

对于partitionBy：

df.rdd.keyBy(lambda x: x.RowKey2).partitionBy(n)

您还错误地假设 partitionBy 像 groupByKey 一样工作，并且您获得单个 RowKey 的项目。不是这样的：How does HashPartitioner work?