如何在Python中检查UDF函数中pyspark数据帧列的单元格值为none或NaN以实现前向填充？

我基本上是在尝试进行前向填充插补。下面是代码。

df = spark.createDataFrame([(1,1, None), (1,2, 5), (1,3, None), (1,4, None), (1,5, 10), (1,6, None)], ('session',"timestamp", "id"))

PRV_RANK = 0.0
def fun(rank):
    ########How to check if None or Nan?  ###############
    if rank is None or rank is NaN:
        return PRV_RANK
    else:
        PRV_RANK = rank
        return rank        

fuN= F.udf(fun, IntegerType())

df.withColumn("ffill_new", fuN(df["id"])).show()

我在日志中收到奇怪的错误。

编辑： 问题与如何使用 python 识别 Spark 数据框中的 null 和 nan 有关。

编辑： 我假设以下检查 NaN 和 Null 的代码行导致了问题。所以我给这个问题起了相应的标题。

回溯（最近一次调用最后一次）：

文件“”，第 1 行，位于 df_na.withColumn("ffill_new",forwardFill(df_na["id"])).show()

文件“C:\Spark\python\pyspark\sql\dataframe.py”，第 318 行，显示 打印（self._jdf.showString（n，20））

文件“C:\Spark\python\lib\py4j-0.10.4-src.zip\py4j\java_gateway.py”， 第 1133 行，在call答案，self.gateway_client，self.target_id，self.name）

文件“C:\Spark\python\pyspark\sql\utils.py”，第 63 行，装饰形式 返回 f(*a, **kw)

文件“C:\Spark\python\lib\py4j-0.10.4-src.zip\py4j\protocol.py”， 第 319 行，在 get_return_value 中 格式（target_id，“。”，名称），值）

Py4JJavaError：调用 o806.showString 时发生错误。 : org.apache.spark.SparkException：作业由于阶段失败而中止： 阶段 47.0 中的任务 0 失败 1 次，最近一次失败：丢失任务 0.0 阶段 47.0（TID 83，本地主机，执行器驱动程序）：org.apache.spark.api.python.PythonException：Traceback（最新 最后调用）：文件 “C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 174 行，在 main 中 文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 169 行，位于 处理文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”， 文件中的第 106 行 “C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 92 行，位于 文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”， 第 70 行，在文件“”第 5 行中， 在forwardfil UnboundLocalError中：引用了局部变量“PRV_RANK” 分配之前

在 org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193) 在 org.apache.spark.api.python.PythonRunner$$anon$1。(PythonRDD.scala:234) 在 org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:144) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:87) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) 在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:99) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:748)

驱动程序堆栈跟踪：位于 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1435) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1423) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1422) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1422) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802) 在 scala.Option.foreach(Option.scala:257) 处 org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:802) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1650) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1605) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1594) 在 org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:628) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1925) 处 org.apache.spark.SparkContext.runJob（SparkContext.scala：1938）在 org.apache.spark.SparkContext.runJob（SparkContext.scala：1951）在 org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:333) 在 org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38) 在 org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1.apply(Dataset.scala:2386) 在 org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57) 在 org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:27​​88) 在 org.apache.spark.sql.Dataset.org$apache$spark$sql$数据集$$执行$1(Dataset.scala:2385) 在 org.apache.spark.sql.Dataset.org$apache$spark$sql$数据集$$收集(Dataset.scala:2392) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2128) 在 org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2127) 在org.apache.spark.sql.Dataset.withTypedCallback（Dataset.scala：2818） 在 org.apache.spark.sql.Dataset.head(Dataset.scala:2127) 处 org.apache.spark.sql.Dataset.take(Dataset.scala:2342) 在 org.apache.spark.sql.Dataset.showString(Dataset.scala:248) 在 sun.reflect.GenerateMethodAccessor35.invoke（来源未知）位于 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 处 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 在 py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 在 py4j.Gateway.invoke(Gateway.java:280) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 处 py4j.GatewayConnection.run(GatewayConnection.java:214) 在 java.lang.Thread.run(Thread.java:748) 原因： org.apache.spark.api.python.PythonException：回溯（最新 最后调用）：文件 “C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 174 行，在 main 中 文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 169 行，位于 处理文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”， 文件中的第 106 行 “C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”，第 92 行，位于 文件“C:\Spark\python\lib\pyspark.zip\pyspark\worker.py”， 第 70 行，在文件“”第 5 行中， 在forwardfil UnboundLocalError中：引用了局部变量“PRV_RANK” 分配之前

在 org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193) 在 org.apache.spark.api.python.PythonRunner$$anon$1。(PythonRDD.scala:234) 在 org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:144) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$doExecute$1.apply(BatchEvalPythonExec.scala:87) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) 在 org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:797) 在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323) 在 org.apache.spark.rdd.RDD.iterator(RDD.scala:287) 处 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87) 在 org.apache.spark.scheduler.Task.run(Task.scala:99) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 还有 1 个

df.withColumn("ffill_new", f.UserDefinedFunction(lambda x: x or 0, IntegerType())(df["id"])).show()