比较Scala中连续行中的列值

Question

我是 Spark Scala 的新手。我遇到了一种情况，我必须比较数据集中特定列的值，例如：

源数据

Source  Destination Distance
Austin  Houston     200
Dallas  Houston     400
Kansas  Dallas      700 

结果

Source1  Destination1   Distance1   Source2  Destination2 Distance2  DistDiff     
Dallas   Houston        400         Kansas   Dallas       700        300

根据情况，我必须比较后续行的距离，如果差异大于或等于 300，则将记录保存在 Resultant 数据集中 700 - 400 = 300

我遇到的示例是具有在任何特定数据集上按行执行的函数，但我的场景是使用连续行。

Answer 1

您提到您可以按日期时间对行进行排序。因此，假设它使用sortBy 或sortByKey 进行排序以创建 ordered rdd，并且假设您有偶数行（因此每一行都有另一行来计算差异），您可以：

1. 使用zipWithIndex为每一行指定一个索引。
2. 通过过滤创建的索引，将 RDD 拆分为两个 RDD，一个具有偶数索引，一个具有奇数索引。
3. zip 将 RDD 拆分在一起，创建一个新的 Tuple2 的 RDD，其中偶数索引行位于左侧，奇数索引行位于右侧。
4. map计算每行左右差的结果。

Answer 2

这可以通过以下方式完成：

1. 向排序的 rdd 添加索引列
2. 确保 rdd 有偶数行 N
3. 创建一个 rdd rdd_even1 以包含索引为 [0, N-2] 的偶数行
4. 创建一个 rdd rdd_odd1 以包含奇数行 [1, N-1]
5. 创建一个 rdd rdd_even2 以包含偶数行 [2, N-2]
6. 创建一个 rdd rdd_odd2 以包含奇数行 [1, N-3]
7. 现在您需要在压缩之前重新分区 rdd_even1 和 *rdd_odd1，因为如果两个 rdd 在所有分区中的元素数量不同（至少在 pyspark 中），压缩将不起作用。您可以使用 collect 和 parallelize 在内存中执行此操作，但很可能您必须将 rdd 写入 HDFS 并重新读取它们，以控制分区
8. 对 rdd_even2 和 rdd_odd2 执行相同的操作
9. 将第 7 步中的 rdd 压缩到 rdd_zip1
10. 将第 8 步中的 rdd 压缩到 rdd_zip2
11. 调用rdd_zip1.union(rdd_zip2)
12. 现在您可以在联合上调用 map() 以获得具有所需差异的“结果”

祝你好运。