Scala/Spark: 如何将值列表转换为单独的行？

Question

我可以帮您进行翻译。以下是所需翻译内容：

我刚接触Scala/Spark，想要将值的列表仅通过RDD转换成单独的行（不使用数据框）。如果有任何人能够帮助我，将不胜感激。

输入：

List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000)) )

预期输出：

A 10643
A 10692
B 10308
C 1000
C 2000

我可以分别完成它们，但无法同时完成。

这是我尝试的方法：

val Input = sc.makeRDD(List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000))  ))

Input.map(value=>value._1).collect().foreach(println)<
A
B
C

Input.map(value=>value._2).flatMap(x=>x).collect().foreach(println)<br>
10643
10692
10308
1000
2000

- ROHAN NAYAK

Answer 1

使用函数 explode 来为给定的数组或映射列中的每个元素创建一个新行。

import org.apache.spark.sql.functions._

val data = List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000)) )
val rdd = sc.parallelize(data) 
val df = rdd.toDF("name", "list")

val exploded = df.withColumn("value", explode($"list")).drop("list")
exploded.show

如果您确实更喜欢使用RDD，请使用以下内容。

val flatted = rdd.flatMap(r => r._2.map((r._1, _)))

- Rockie Yang

Answer 2

这里提供一种与Spark无关的解决方案：

val list = List(("A", List(10643, 10692)), ("B", List(10308)), ("C", List(1000, 2000)))

val result = list.flatMap { 
  case (key, value) => value.map(v => (key, v))
}

result.foreach(println)

// (A,10643)
// (A,10692)
// (B,10308)
// (C,1000)
// (C,2000)

- slouc

Answer 3

使用PairRDDFunctions

import scala.collection.mutable._
val df = List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000)) ).toDF("name","list")
val rdd1 = df.rdd.map( x=>  (x(0), x(1)))
val pair = new PairRDDFunctions(rdd1)
pair.flatMapValues(_.asInstanceOf[mutable.WrappedArray[Int]]).foreach(println)

结果：

(A,10643)
(B,10308)
(A,10692)
(C,1000)
(C,2000)

- stack0114106

Answer 4

两种解决方案（第二个更易扩展）：

在驱动程序上进行处理，然后转换为RDD

val in = List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000)) )

val out = sc.parallelize(in.flatMap{case (k, l) => l.map(v => (k,v))})

out.take(10).foreach(println)

将数据转换为RDD，然后在执行器上进行处理

val in = List( ("A",List(10643, 10692)), ("B",List(10308)),("C",List(1000,2000)) )

val inRDD = sc.parallelize(in)

val out = inRDD.flatMap{case (k, l) => l.map(v => (k,v))}

- Ben Horsburgh