星火數據框 - 從給定一個數據幀

Question

列

下降空值：

val df = sc.parallelize(Seq(("foo", ArrayBuffer(null,"bar",null)), ("bar", ArrayBuffer("one","two",null)))).toDF("key", "value") 
    df.show 

    +---+--------------------------+ 
    |key|      value| 
    +---+--------------------------+ 
    |foo|ArrayBuffer(null,bar,null)| 
    |bar|ArrayBuffer(one, two,null)| 
    +---+--------------------------+ 

我想從value列拖放null。刪除後，數據幀應如下所示：

+---+--------------------------+ 
    |key|      value| 
    +---+--------------------------+ 
    |foo|ArrayBuffer(bar)   | 
    |bar|ArrayBuffer(one, two)  | 
    +---+--------------------------+ 

任何建議歡迎。 10x

Answer 1

這裏你需要一個UDF。例如用flatMap：

val filterOutNull = udf((xs: Seq[String]) => 
    Option(xs).map(_.flatMap(Option(_)))) 

df.withColumn("value", filterOutNull($"value")) 

在外部Option與map處理NULL列：

Option(null: Seq[String]).map(identity) 

Option[Seq[String]] = None 

Option(Seq("foo", null, "bar")).map(identity) 

Option[Seq[String]] = Some(List(foo, null, bar)) 

，並確保當輸入是我們不失敗，NPE NULL/null按地圖平

NULL -> null -> None -> None -> NULL 

其中null是斯卡拉null和NULL是SQL NULL。

內部flatMap變平的Options序列有效過濾nulls：

Seq("foo", null, "bar").flatMap(Option(_)) 

Seq[String] = List(foo, bar) 

更迫切當量可以是這樣的：

val imperativeFilterOutNull = udf((xs: Seq[String]) => 
    if (xs == null) xs 
    else for { 
    x <- xs 
    if x != null 
    } yield x) 

Answer 2

選項1：使用UDF：

val filterNull = udf((arr : Seq[String]) => arr.filter((x: String) => x != null)) 
df.withColumn("value", filterNull($"value")).show() 

選項2：沒有UDF

df.withColumn("value", explode($"value")).filter($"value".isNotNull).groupBy("key").agg(collect_list($"value")).show() 

注意，這是效率不高......