根据另一个 RDD Spark (Java) 的值聚合一个 RDD

Question

我有两个包含时间信息的 RDD。 RDD 被分成不同的分区。 一种是形式

16:00:00
16:00:18
16:00:25
16:01:01
16:01:34
16:02:12
16:02:42
...

和另一个包含 tuple2 形式的时间跨度

<16:00:00, 16:00:59>
<16:01:00, 16:01:59>
<16:02:00, 16:02:59>
...

我需要聚合第一个和第二个 RDD，通过根据第二个中的值聚合第一个 RDD 的值，以获得类似的东西

<<16:00:00, 16:00:59>, [16:00:00,16:00:18,16:00:25]>
<<16:01:00, 16:01:59>, [16:01:01,16:01:34]>
<<16:02:00, 16:02:59>, [16:02:12,16:02:42]>
...

或者，或者，类似的东西

<<16:00:00, 16:00:59>, 16:00:00>
<<16:00:00, 16:00:59>, 16:00:18>
<<16:00:00, 16:00:59>, 16:00:25>
<<16:01:00, 16:01:59>, 16:01:01>
<<16:01:00, 16:01:59>, 16:01:34>
<<16:02:00, 16:02:59>, 16:02:12>
<<16:02:00, 16:02:59>, 16:02:42>
...

我正在尝试使用整个范围的 spark 转换函数，但我很难找到一个适用于这种不同性质的 RDD 的函数。我知道我可能会选择cartesian 产品，然后进行过滤，但我想要一个“更好”的解决方案。我试过zipPartition，这可能有效，但我的分区可能不一致，例如16:00:00 可能最终出现在不存在相应聚合值（元组<16:00:00, 16:00:59>）的分区中。 解决这个问题的最佳方法是什么？

PS：我使用的是 Java，但也欢迎使用 Scala 解决方案。 谢谢

Answer 1

我已将以下内容简化为使用整数，但我相信同样可以多次使用。虽然示例是用 Scala 编写的，但我怀疑它也可以用 Java 完成。

如果范围是常规的，我会将“值”RDD 转换为 range,value，然后进行简单的连接。

val values = Seq(1, 5, 10, 14, 20)
val valuesRdd = sc.parallelize(values, 2)
valuesRdd.map(x => (((x/10)*10, ((x/10)*10)+9), x)).collect

但是，如果范围不规则，则：

如果您不介意使用 DataFrames，那么一个选项是使用 用户定义的函数 根据是否 V 在给定范围内创建列并加入该列。

case class Range(low : Int, high :Int)
val ranges = Seq( Range(0,9), Range(10,19), Range(20,29));
val rangesDf = sc.parallelize(ranges, 2).toDF

case class Value(value : Int)
val values = Seq(Value(1), Value(5), Value(10), Value(14), Value(20))
val valuesDf = sc.parallelize(values, 2).toDF

val inRange = udf{(v: Int, low: Int, high : Int) => v >= low && v<= high}

rangesDf.join(valuesDf, inRange(valuesDf("value"), rangesDf("low"), rangesDf("high"))).show

下一个选项是分解范围并加入分解后的版本：

val explodedRange = rangesRdd.map(x => (x, List.range(x._1, x._2 + 1))).flatMap( { case (range, lst) => lst.map { x => (x, range)} })
val valuesRdd = sc.parallelize(values, 2).map(x => (x,true))
valuesRdd.join(explodedRange).map(x => (x._2._2, x._1)).collect