在 spark 数据框中使用列值转换另一列

Posted 2023-04-17

【中文标题】在 spark 数据框中使用列值转换另一列

【英文标题】：Using a columns value in casting another column in a spark dataframe

【发布时间】：2018-08-27 21:04:46

【问题描述】：

我有一个这样的数据框：

rdd1 = sc.parallelize([(100,2,1234.5678),(101,3,1234.5678)])
df = spark.createDataFrame(rdd1,(['id','dec','val']))

+---+---+---------+
| id|dec|      val|
+---+---+---------+
|100|  2|1234.5678|
|101|  3|1234.5678|
+---+---+---------+

根据dec 列中可用的值，我希望在val 列上进行转换。就像dec = 2 一样，我希望将val 转换为DecimalType(7,2)。

我尝试执行以下操作，但它不起作用：

 df.select(col('id'),col('dec'),col('val'),col('val').cast(DecimalType(7,col('dec'))).cast(StringType()).alias('modVal')).show()

错误信息：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/spark/python/pyspark/sql/column.py", line 419, in cast
    jdt = spark._jsparkSession.parseDataType(dataType.json())
  File "/usr/lib/spark/python/pyspark/sql/types.py", line 69, in json
    return json.dumps(self.jsonValue(),
  File "/usr/lib/spark/python/pyspark/sql/types.py", line 225, in jsonValue
    return "decimal(%d,%d)" % (self.precision, self.scale)
TypeError: %d format: a number is required, not Column

如果我将值硬编码为一个特定的数字，这同样有效。

df.select(col('id'),col('dec'),col('val'),col('val').cast(DecimalType(7,3)).cast(StringType()).alias('modVal')).show()

+---+---+---------+--------+
| id|dec|      val|  modVal|
+---+---+---------+--------+
|100|  2|1234.5678|1234.568|
|101|  3|1234.5678|1234.568|
+---+---+---------+--------+

请帮帮我。

【参考方案1】：

Spark（或任何相关系统）中的列必须是同质的 - 像这样的操作，您 cast 将每一行转换为不同的类型，不仅不受支持，而且没有多大意义。

【讨论】：

可能是我在这里遗漏了一些东西。但是，请帮助我理解为什么不支持/无效地转换 spark 数据框中的每一列的行。我明白了，根据其他列值强制转换列可能是一个不受欢迎的用例，但不知道为什么你会说整个想法没有任何意义。

@vishnuram：同一列中所有行的数据类型必须相同。但是，如果您只是在格式化之后，您可以在这种情况下使用字符串，这将使数据类型相同，同时允许不同的小数位数。

@shaido & user10281832 谢谢。现在我明白了为什么人们担心数据类型是异构的，并更新了我的命令，将其转换为 StringType() 以使我的请求更加清晰

【参考方案2】：

正如 user10281832 所述，同一列中不能有不同的数据类型。

由于格式化是焦点，您可以将列转换为字符串类型，然后进行格式化。由于每行的小数位数不同，因此不能使用任何内置的 Spark 函数，但需要定义自定义 UDF：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def format_val(num, prec):
    return "%0.*f" % (prec, num)

format_val_udf = udf(format_val, StringType())

df.withColumn('modVal', format_val_udf('val', 'dec'))