相关疑难解决方法(0)

我有一个类型DataFrame很多的列str,我想将一个函数应用于所有这些列,而不重命名它们或添加更多列,我尝试使用for-in循环执行withColumn(参见下面的示例),但通常在我运行代码时,它显示了Stack Overflow(它很少有效),这DataFrame根本不大,它只有~15000条记录.

# df is a DataFrame
def lowerCase(string):
    return string.strip().lower()

lowerCaseUDF = udf(lowerCase, StringType())

for (columnName, kind) in df.dtypes:
    if(kind == "string"):
        df = df.withColumn(columnName, lowerCaseUDF(df[columnName]))

df.select("Tipo_unidad").distinct().show()

完整的错误很长,因此我决定只粘贴一些行.但是你可以在这里找到完整的跟踪

Py4JJavaError:调用o516.showString时发生错误.:org.apache.spark.SparkException:作业因阶段失败而中止:阶段2.0中的任务1失败4次,最近失败:阶段2.0中丢失的任务1.3(TID 38,worker2.mcbo.mood.com.ve): java.io.ObjectInputStream中的java.lang.StackOverflowError $ BlockDataInputStream.readByte(ObjectInputStream.java:2774)

我认为这个问题产生的原因是这个代码启动了许多工作(每个类型一个string),你能告诉我另一个选择或我做错了吗？

我在使用Pyspark Dataframe时遇到了一些麻烦.具体来说,我正在尝试为数据帧创建一个列,这是合并数据帧的两列的结果.

例如

this_dataframe = this_dataframe.withColumn('new_max_price', coalesce(this_dataframe['max_price'],this_dataframe['avg(max_price)']).cast(FloatType()))

此代码的问题是它仍然在某些行中返回值"null".具体来说,我正在运行此代码:

this_dataset.where(col("new_max_price").isNull()).count()

此代码给出了积极的结果.因此,虽然此代码有效,但它不会产生预期的结果.

我发现了一些其他问题(例如从PySpark DataFrame中的非空列中选择值),这些问题被认为是相似的,但由于某种原因,我无法复制他们的结果.

这里有一些基于前面提到的链接的代码:

def coalesce_columns(c1, c2):
    if c1 != None and c2 != None:
        return c1
    elif c1 == None:
        return c2
    else:
        return c1

coalesceUDF = udf(coalesce_columns)
max_price_col = [coalesceUDF(col("max_price"), col("avg(max_price)")).alias("competitive_max_price")]
this_dataset.select(max_price_col).show()

当我尝试执行最后一行来测试我的结果是否正确时,我收到一个错误.

AttributeError:'unicode'对象没有属性'isNull'

基本上问题是,如何使用spark sql函数创建一个合并两个pyspark数据帧列的列？如果这是不可能的,我可以使用什么样的UDF来创建一些我可以附加到另一个数据帧的数据帧列？