Spark 爆炸/posexplode 列值 [英] Spark explode/posexplode column value
本文介绍了Spark 爆炸/posexplode 列值的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我对 spark 非常陌生，我想以这样的方式分解我的 df，它会创建一个带有拆分值的新列，并且它还具有该特定值与其行相关的顺序或索引.
I am very new to spark and I want to explode my df in such a way that it will create a new column with its splited values and it also has the order or index of that particular value respective to its row.
CODE:
import spark.implicits._
val df = Seq("40000.0~0~0~", "0~40000.0~", "0~", "1000.0~0~0~", "1333.3333333333333~0~0~0~0", "66666.66666666667~0~0~")
.toDF("VALUES")
df.show(false)
Input DF:
+--------------------------+
|VALUES |
+--------------------------+
|40000.0~0~0~ |
|0~40000.0~ |
|0~ |
|1000.0~0~0~ |
|1333.3333333333333~0~0~0~0|
|66666.66666666667~0~0~ |
+--------------------------+
Output DF:
+------+------------------+-----------+
|row_id|col |order |
+------+------------------+-----------+
|1 |40000.0 |1 |
|1 |0 |2 |
|1 |0 |3 |
|1 | |4 |<== don't want this column with empty or null value
|2 |0 |1 |
|2 |40000.0 |2 |
|2 | |3 |<== don't want this column with empty or null value
|3 |0 |1 |
|3 | |2 |<== don't want this column with empty or null value
|4 |1000.0 |1 |
|4 |0 |2 |
|4 |0 |3 |
|4 | |4 |<== don't want this column with empty or null value
|5 |1333.3333333333333|1 |
|5 |0 |2 |
|5 |0 |3 |
|5 |0 |4 |
|5 |0 |5 |
|6 |66666.66666666667 |1 |
|6 |0 |2 |
|6 |0 |3 |
|6 | |4 |<== don't want this column with empty or null value
+------+------------------+-----------+
也不希望此列具有空值或空值.
Also don't want this column with an empty or null value.
如何在 Scala - spark 中完成此操作?
How this can be done in scala - spark?
推荐答案
使用window函数添加row_id然后使用posexplode 和 filter 过滤掉空值.
Use window function to add row_id then use posexplode and filter to filter out the empty value.
示例:
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._
val w=Window.orderBy(col("id"))
df.withColumn("id",monotonically_increasing_id()).
withColumn("row_id",row_number().over(w)).
selectExpr("row_id","posexplode(split(values,'~')) as (pos, val)").
withColumn("order",col("pos") + 1).
drop("pos").
filter(length(col("val")) !== 0).
show()
//or using expr
df.withColumn("id",monotonically_increasing_id()).
withColumn("row_id",row_number().over(w)).
withColumn("arr",expr("filter(split(values,'~'),x -> x != '')")).
selectExpr("row_id","""posexplode(arr) as (pos, val)""").
withColumn("order",col("pos") + 1).
drop("pos").
show()
//+------+------------------+-----+
//|row_id| val|order|
//+------+------------------+-----+
//| 1| 40000.0| 1|
//| 1| 0| 2|
//| 1| 0| 3|
//| 2| 0| 1|
//| 2| 40000.0| 2|
//| 3| 0| 1|
//| 4| 1000.0| 1|
//| 4| 0| 2|
//| 4| 0| 3|
//| 5|1333.3333333333333| 1|
//| 5| 0| 2|
//| 5| 0| 3|
//| 5| 0| 4|
//| 5| 0| 5|
//| 6| 66666.66666666667| 1|
//| 6| 0| 2|
//| 6| 0| 3|
//+------+------------------+-----+
<小时>
从Spark-2.4+我们可以使用array_remove函数来过滤""
df.withColumn("id",monotonically_increasing_id()).
withColumn("row_id",row_number().over(w)).
selectExpr("row_id","posexplode(array_remove(split(values,'~'),'')) as (pos, val)").
withColumn("order",col("pos") + 1).
drop("pos").
show()
//+------+------------------+-----+
//|row_id| val|order|
//+------+------------------+-----+
//| 1| 40000.0| 1|
//| 1| 0| 2|
//| 1| 0| 3|
//| 2| 0| 1|
//| 2| 40000.0| 2|
//| 3| 0| 1|
//| 4| 1000.0| 1|
//| 4| 0| 2|
//| 4| 0| 3|
//| 5|1333.3333333333333| 1|
//| 5| 0| 2|
//| 5| 0| 3|
//| 5| 0| 4|
//| 5| 0| 5|
//| 6| 66666.66666666667| 1|
//| 6| 0| 2|
//| 6| 0| 3|
//+------+------------------+-----+
这篇关于Spark 爆炸/posexplode 列值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
