拆分单行的多个字段或列并使用 Scala 创建多行答案

【问题标题】：Split multiple fields or columns of a single row and create multiple rows using Scala拆分单行的多个字段或列并使用 Scala 创建多行
【发布时间】：2020-05-16 17:00:01
【问题描述】：

我有一个包含 4 个字段的数据框，如下所述：

Field1 , Field2 , Field3 , Field4

我在以下字段中有值：

A1      , B1         , C1         , D1                
A2      , B2,B3      , C2,C3      , D2,D3             
A1      , B4,B5,B6   , C4,C5,C6   , D4,D5,D6

我必须将其转换为以下格式：

A1      , B1         , C1         , D1          
A2      , B2         , C2         , D2            
A2      , B3         , C3         , D3      
A1      , B4         , C4         , D4      
A1      , B5         , C5         , D5      
A1      , B6         , C6         , D6

基本上我必须将逗号分隔的值拆分为多列，并根据相同顺序的值形成新行。

您可以将它们全部视为 String 类型。您能否建议我一种方法来根据新值进行拆分并形成新行。

我已经看到了一个与下面类似的问题：

如何在 Spark 中对嵌套的 Dataframe 进行平面映射

但是这个问题是不同的，因为在这种情况下我必须考虑拆分多个列并且值不应该重复。

【问题讨论】：

标记的重复问题与此不同。我这里有多个列要同时拆分。

标签： scala apache-spark dataframe

【解决方案1】：

您可以将DataFrame 转换为Dataset[(String, String, String, String)] 和flatMap：

import scala.util.Try

val df = Seq(
  ("A1", "B1", "C1", "D1"),
  ("A2", "B2,B3", "C2,C3", "D2,D3"),
  ("A1", "B4,B5,B6", "C4,C5,C6", "D4,D5,D6")
).toDF("x1", "x2", "x3", "x4")

// A simple sequence of expressions which allows us to flatten the results
val exprs = (0 until df.columns.size).map(i => $"value".getItem(i))

df.select($"x1", array($"x2", $"x3", $"x4")).as[(String, Seq[String])].flatMap {
  case (x1, xs) => 
    Try(xs.map(_.split(",")).transpose).map(_.map("x" +: _)).getOrElse(Seq())
}.toDF.select(exprs:_*)

// +--------+--------+--------+--------+
// |value[0]|value[1]|value[2]|value[3]|
// +--------+--------+--------+--------+
// |      A1|      B1|      C1|      D1|
// |      A2|      B2|      C2|      D2|
// |      A2|      B3|      C3|      D3|
// |      A1|      B4|      C4|      D4|
// |      A1|      B5|      C5|      D5|
// |      A1|      B6|      C6|      D6|
// +--------+--------+--------+--------+

或使用 UDF：

val splitRow = udf((xs: Seq[String]) => 
   Try(xs.map(_.split(",")).transpose).toOption)

// Same as before but we exclude the first column
val exprs = (0 until df.columns.size - 1).map(i => $"xs".getItem(i))

df
  .withColumn("xs", explode(splitRow(array($"x2", $"x3", $"x4"))))
  .select($"x1" +: exprs: _*)

【讨论】：

太棒了！您的解决方案很好，适用于给定的输入。但是，如果我将输入更改如下：val df = Seq( ("A2", "B2,B3", "C2,C3", "D2,D3"), ).toDF("x1", "x2", "x3", "x4") 它会失败并在转置中给出错误。你能给我一个通用的解决方案吗？
你能提供追溯吗？所有字符串必须具有相同的大小才能使其正常工作。
太棒了！它工作得如此流畅。希望你喜欢回答它。 ;)

【解决方案2】：

您可以使用poseexplode 快速解决此问题。参考http://allabouthadoop.net/hive-lateral-view-explode-vs-posexplode/ 因此，您的代码将如下所示：

select 
    Field1, 
    Field2,
    Field3,
    Field4 
from temp_table
lateral view posexplode(Field2) pn as f2_1,f2_2, Field2
lateral view posexplode(Field3) pn as f3_1,f3_2, Field3 
lateral view posexplode(Field3) pn as f4_1,f4_2, Field4
where 
    (f2_1 == F3_1 and f3_1 == f4_1) and/or (f2_2 == F3_2 and f3_2 == f4_2)

【讨论】：