use*_*787 7 scala apache-spark apache-spark-dataset
我面临的问题是如何拆分多值列,即List[String]分成不同的行.
初始数据集具有以下类型: Dataset[(Integer, String, Double, scala.List[String])]
+---+--------------------+-------+--------------------+
| id| text | value | properties |
+---+--------------------+-------+--------------------+
| 0|Lorem ipsum dolor...| 1.0|[prp1, prp2, prp3..]|
| 1|Lorem ipsum dolor...| 2.0|[prp4, prp5, prp6..]|
| 2|Lorem ipsum dolor...| 3.0|[prp7, prp8, prp9..]|
Run Code Online (Sandbox Code Playgroud)
生成的数据集应具有以下类型:
Dataset[(Integer, String, Double, String)]
Run Code Online (Sandbox Code Playgroud)
而properties应拆分这样的:
+---+--------------------+-------+--------------------+
| id| text | value | property |
+---+--------------------+-------+--------------------+
| 0|Lorem ipsum dolor...| 1.0| prp1 |
| 0|Lorem ipsum dolor...| 1.0| prp2 |
| 0|Lorem ipsum dolor...| 1.0| prp3 |
| 1|Lorem ipsum dolor...| 2.0| prp4 |
| 1|Lorem ipsum dolor...| 2.0| prp5 |
| 1|Lorem ipsum dolor...| 2.0| prp6 |
Run Code Online (Sandbox Code Playgroud)
explode经常被建议,但它来自无类型的DataFrame API,并且假设您使用数据集,我认为flatMap运算符可能更合适(请参阅org.apache.spark.sql.Dataset).
flatMap[U](func: (T) ? TraversableOnce[U])(implicit arg0: Encoder[U]): Dataset[U]
Run Code Online (Sandbox Code Playgroud)
(特定于Scala)通过首先将函数应用于此数据集的所有元素,然后展平结果,返回新的数据集.
您可以按如下方式使用它:
val ds = Seq(
(0, "Lorem ipsum dolor", 1.0, Array("prp1", "prp2", "prp3")))
.toDF("id", "text", "value", "properties")
.as[(Integer, String, Double, scala.List[String])]
scala> ds.flatMap { t =>
t._4.map { prp =>
(t._1, t._2, t._3, prp) }}.show
+---+-----------------+---+----+
| _1| _2| _3| _4|
+---+-----------------+---+----+
| 0|Lorem ipsum dolor|1.0|prp1|
| 0|Lorem ipsum dolor|1.0|prp2|
| 0|Lorem ipsum dolor|1.0|prp3|
+---+-----------------+---+----+
// or just using for-comprehension
for {
t <- ds
prp <- t._4
} yield (t._1, t._2, t._3, prp)
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
3192 次 |
| 最近记录: |