spark explode/posexplode列值

vdgimpew 于 2021-05-29 发布在 Spark

关注(0)|答案(2)|浏览(703)

我对spark非常陌生，我想以这样一种方式分解df，它将创建一个新列，其中包含拆分的值，并且它还具有该特定值对应于其行的顺序或索引。

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
+------+------------------+-----------+

也不希望此列具有空值或空值。
如何在scala-spark中实现这一点？

apache-spark apache-spark-sql

来源：https://stackoverflow.com/questions/62337056/spark-explode-posexplode-column-value

2条答案

按热度按时间

1tuwyuhd1#

您需要过滤空/空值。这就是你需要做的

from pyspark.sql.types import *
from pyspark.sql.functions import *
value1 = Row(VALUES='40000.0~0~0~')
value2 = Row(VALUES='0~40000.0~')
value3 = Row(VALUES='1000.0~0~0~')
value4 = Row(VALUES='333.3333333333333~0~0~0~0')
value5 = Row(VALUES='66666.66666666667~0~0~')

schema = StructType([StructField('VALUES', StringType())])
rows = [value1,value2,value3,value4,value5]
df = spark.createDataFrame(rows, schema)

df1= df.withColumn("newCol",explode(split("VALUES","~")))

df1=df1.filter(length("newCol")!=0)

df1.show(50,False)

这是输出

+-------------------------+-----------------+
|VALUES                   |newCol           |
+-------------------------+-----------------+
|40000.0~0~0~             |40000.0          |
|40000.0~0~0~             |0                |
|40000.0~0~0~             |0                |
|0~40000.0~               |0                |
|0~40000.0~               |40000.0          |
|1000.0~0~0~              |1000.0           |
|1000.0~0~0~              |0                |
|1000.0~0~0~              |0                |
|333.3333333333333~0~0~0~0|333.3333333333333|
|333.3333333333333~0~0~0~0|0                |
|333.3333333333333~0~0~0~0|0                |
|333.3333333333333~0~0~0~0|0                |
|333.3333333333333~0~0~0~0|0                |
|66666.66666666667~0~0~   |66666.66666666667|
|66666.66666666667~0~0~   |0                |
|66666.66666666667~0~0~   |0                |
+-------------------------+-----------------+

赞(0）回复(0）举报 2021-05-29

ykejflvf2#

使用窗口功能添加 row_id 然后使用 posexplode 和filter以过滤掉空值。 Example: ```
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|
//+------+------------------+-----+

赞(0）回复(0）举报 2021-05-29

我来回答

spark explode/posexplode列值

2条答案

相关问题

热门标签

最新问答