展平火花scala中的元素答案

【问题标题】：Flatten an element in spark scala展平火花scala中的元素
【发布时间】：2015-10-19 14:51:25
【问题描述】：

我在 RDD 中有这样的数据：

RDD[((Int, Int, Int), ((Int, Int), Int))]

作为：

(((9,679,16),((2,274),1)), ((250,976,13),((2,218),1)))

我想输出为：

((9,679,16,2,274,1),(250,976,13,2,218,1))

加入 2 个 rdds 后：

val joinSale = salesTwo.join(saleFinal)

我得到了那个结果集。我尝试了以下代码。

joinSale.flatMap(x => x).take(100).foreach(println)

我尝试过 map/flatMap 但做不到。任何想法如何实现这样的场景？提前谢谢..

【问题讨论】：

Tuple 没有 flatten 方法。但是您可以使用productIterator 将其展平 - stackoverflow.com/questions/5289408/iterate-over-a-tuple 但是这真的很难看。

标签： scala apache-spark rdd

【解决方案1】：

你可以在 scala 中使用模式匹配来做到这一点。只需将您的元组修改逻辑包装在类似于以下的映射中：

val mappedJoinSale = joinSale.map { case ((a, b, c), ((d, e), f)) => (a, b, c, d, e, f) }

使用您的示例，我们有：

scala> val example = sc.parallelize(Array(((9,679,16),((2,274),1)), ((250,976,13),((2,218),1))))
example: org.apache.spark.rdd.RDD[((Int, Int, Int), ((Int, Int), Int))] = ParallelCollectionRDD[0] at parallelize at <console>:12

scala> val mapped = example.map { case ((a, b, c), ((d, e), f)) => (a, b, c, d, e, f) }
mapped: org.apache.spark.rdd.RDD[(Int, Int, Int, Int, Int, Int)] = MappedRDD[1] at map at <console>:14

scala> mapped.take(2).foreach(println)
...
(9,679,16,2,274,1)
(250,976,13,2,218,1)

【讨论】：

哇，现在我觉得自己很愚蠢..我确实在不久前实施了地图案例..我可能已经忘记了..无论如何谢谢..
@Achillies57 别担心，我们都忘记了事情 :)

【解决方案2】：

您还可以使用奇妙的shapeless 库创建通用元组展平器，如下所示：

import shapeless._
import shapeless.ops.tuple
trait LowLevelFlatten extends Poly1 {
  implicit def anyFlat[T] = at[T](x => Tuple1(x))
}

object concat extends Poly2 {
  implicit def atTuples[T1, T2](implicit prepend: tuple.Prepend[T1, T2]): Case.Aux[T1, T2, prepend.Out] =
    at[T1,T2]((t1,t2) => prepend(t1,t2))
}

object flatten extends LowLevelFlatten {
  implicit def tupleFlat[T, M](implicit
                               mapper: tuple.Mapper.Aux[T, flatten.type, M],
                               reducer: tuple.LeftReducer[M, concat.type]
                                ): Case.Aux[T, reducer.Out] =
    at[T](t => reducer(mapper(t)))
}

现在在import shapeless._ 存在的任何代码中，您都可以将其用作

joinSale.map(flatten)

【讨论】：