【问题标题】:PySpark - Explode columns into rows and set values based on logicPySpark - 将列分解为行并根据逻辑设置值
【发布时间】:2020-03-25 16:05:03
【问题描述】:

给定一个数据框:

+---+-----------+---------+-------+------------+
| id|      score|tx_amount|isValid|    greeting|
+---+-----------+---------+-------+------------+
|  1|        0.2|    23.78|   true| hello_world|
|  2|        0.6|    12.41|  false|byebye_world|
+---+-----------+---------+-------+------------+

我想将这些列分解为名为“col_value”的行。这部分很好,但我也想对每一行应用逻辑,以便得到如下结果:

+---+------------+--------+---------+----------+-------+
| id|   col_value|is_score|is_amount|is_boolean|is_text|
+---+------------+--------+---------+----------+-------+
|  1|         0.2|       Y|        N|         N|      N|
|  1|       23.78|       N|        Y|         N|      N|
|  1|        true|       N|        N|         Y|      N|
|  1| hello_world|       N|        N|         N|      Y|
|  2|         0.6|       Y|        N|         N|      N|
|  2|       12.41|       N|        Y|         N|      N|
|  2|       false|       N|        N|         Y|      N|
|  2|byebye_world|       N|        N|         N|      Y|
+---+------------+--------+---------+----------+-------+

到目前为止我所拥有的:

.withColumn("cols", F.explode(F.arrays_zip(F.array("score", "tx_amount", "isValid", "greeting")))) \
        .select("id", F.col("cols.*")) \
        .withColumnRenamed("0", "col_value") \
        .withColumn("is_score", F.lit("Y") if col1_type == "score" else F.lit("N")) \
        .withColumn("is_amount", F.lit("Y") if col2_type == "amount" else F.lit("N")) \
        .withColumn("is_boolean", F.lit("Y") if col3_type == "boolean" else F.lit("N")) \
        .withColumn("is_text", F.lit("Y") if col4_type == "text" else F.lit("N")) \
        .show()

但它给出了错误的输出,因为它为每一列给出了相同的结果:

+---+------------+--------+---------+----------+-------+
| id|   col_value|is_score|is_amount|is_boolean|is_text|
+---+------------+--------+---------+----------+-------+
|  1|         0.2|       Y|        Y|         Y|      Y|
|  1|       23.78|       Y|        Y|         Y|      Y|
|  1|        true|       Y|        Y|         Y|      Y|
|  1| hello_world|       Y|        Y|         Y|      Y|
|  2|         0.6|       Y|        Y|         Y|      Y|
|  2|       12.41|       Y|        Y|         Y|      Y|
|  2|       false|       Y|        Y|         Y|      Y|
|  2|byebye_world|       Y|        Y|         Y|      Y|
+---+------------+--------+---------+----------+-------+

爆炸后我怎样才能得到正确的结果?

【问题讨论】:

  • df.dtypes 为您的原始数据帧返回什么?

标签: python apache-spark pyspark explode


【解决方案1】:

我认为你想要的可以通过在你的col_value 上应用regex 来确定它是否是text,boolean,amount or score 来实现。只要score 从不超过1.0,并且amount 始终高于1.0,下面的代码就可以工作。如果不是这种情况,请告诉我我将更新逻辑。

from pyspark.sql import functions as F
df.withColumn("cols", F.explode(F.arrays_zip(F.array("score", "tx_amount", "isValid", "greeting")))) \
        .select("id", F.col("cols.*")) \
        .withColumnRenamed("0", "col_value")\
        .withColumn("text", (F.regexp_extract(F.col("col_value"),"([A-Za-z]+)",1)))\
        .withColumn("boolean", F.when((F.col("text")=='true')|(F.col("text")=='false'),F.col("text")).otherwise(F.lit("")))\
        .withColumn("text", F.when(F.col("text")==F.col("boolean"), F.lit("")).otherwise(F.col("text")))\
        .withColumn("numeric", F.regexp_extract(F.col("col_value"),"([0-9]+)",1))\
        .withColumn("is_text", F.when(F.col("text")!="", F.lit("Y")).otherwise(F.lit("N")))\
        .withColumn("is_score", F.when(F.col("numeric")<=1, F.lit("Y")).otherwise(F.lit("N")))\
        .withColumn("is_amount", F.when(F.col("numeric")>1, F.lit("Y")).otherwise(F.lit("N")))\
        .withColumn("is_boolean", F.when(F.col("boolean")!="", F.lit("Y")).otherwise(F.lit("N")))\
        .select("id", "col_value","is_score","is_amount","is_boolean","is_text").show()


+---+------------+--------+---------+----------+-------+
| id|   col_value|is_score|is_amount|is_boolean|is_text|
+---+------------+--------+---------+----------+-------+
|  1|         0.2|       Y|        N|         N|      N|
|  1|       23.78|       N|        Y|         N|      N|
|  1|        true|       N|        N|         Y|      N|
|  1| hello_world|       N|        N|         N|      Y|
|  2|         0.6|       Y|        N|         N|      N|
|  2|       12.41|       N|        Y|         N|      N|
|  2|       false|       N|        N|         Y|      N|
|  2|byebye_world|       N|        N|         N|      Y|
+---+------------+--------+---------+----------+-------+

【讨论】:

  • 如果您知道输入数据类型是否也可以这样做:例如 [int, float, boolean, string] ?如果我使用这样的数据类型,是否可以以某种方式将其压缩到:F.arrays_zip(F.array("score", "tx_amount", "isValid", "greeting"))
  • @bp2010 不确定它是否适用于您的情况,因为您将多个数据类型压缩并分解到同一列,但可能有一种解决方法。如果您可以提供有关您要查找的内容的更多详细信息,建议您打开一个新问题
  • 为此创建了新问题:stackoverflow.com/questions/61079906/…
  • spark 2.3 中 F.arrays_zip 的任何替代品?
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2021-11-12
  • 1970-01-01
  • 1970-01-01
  • 2019-03-23
  • 2018-04-29
  • 2022-11-11
相关资源
最近更新 更多