自定义模块中的函数在 PySpark 中不起作用，但在交互模式下输入时它们起作用
Functions from custom module not working in PySpark, but they work when inputted in interactive mode
我有一个我编写的模块，其中包含作用于 PySpark DataFrames 的函数。他们对 DataFrame 中的列进行转换，然后 return 一个新的 DataFrame。这是代码示例，已缩短为仅包含其中一个功能：
from pyspark.sql import functions as F
from pyspark.sql import types as t
import pandas as pd
import numpy as np
metadta=pd.DataFrame(pd.read_csv("metadata.csv")) # this contains metadata on my dataset
def str2num(text):
if type(text)==None or text=='' or text=='NULL' or text=='null':
return 0
elif len(text)==1:
return ord(text)
else:
newnum=''
for lettr in text:
newnum=newnum+str(ord(lettr))
return int(newnum)
str2numUDF = F.udf(lambda s: str2num(s), t.IntegerType())
def letConvNum(df): # df is a PySpark DataFrame
#Get a list of columns that I want to transform, using the metadata Pandas DataFrame
chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist()
for curcol in chng_cols:
df=df.withColumn(curcol, str2numUDF(df[curcol]))
return df
这就是我的模块，称之为 mymodule.py。如果我启动 PySpark shell，然后执行以下操作：
import mymodule as mm
myf=sqlContext.sql("select * from tablename lim 10")
我检查了 myf (PySpark DataFrame)，没问题。我通过尝试使用 str2num 函数检查我是否实际导入了 mymodule：
mm.str2num('a')
97
所以它实际上是在导入模块。那么如果我试试这个：
df2=mm.letConvNum(df)
并执行此操作以检查它是否有效：
df2.show()
它尝试执行操作，但随后崩溃：
16/03/10 16:10:44 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 365)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
16/03/10 16:10:44 ERROR TaskSetManager: Task 0 in stage 1.0 failed 1 times; aborting job
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/dataframe.py", line 256, in show
print(self._jdf.showString(n, truncate))
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/utils.py", line 36, in deco
return f(*a, **kw)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o7299.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 365, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage.apply(DAGScheduler.scala:1271)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage.apply(DAGScheduler.scala:1270)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed.apply(DAGScheduler.scala:697)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed.apply(DAGScheduler.scala:697)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)
at org.apache.spark.util.EventLoop$$anon.run(EventLoop.scala:48)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1824)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1837)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1850)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:215)
at org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:207)
at org.apache.spark.sql.DataFrame$$anonfun$collect.apply(DataFrame.scala:1385)
at org.apache.spark.sql.DataFrame$$anonfun$collect.apply(DataFrame.scala:1385)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:1903)
at org.apache.spark.sql.DataFrame.collect(DataFrame.scala:1384)
at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1314)
at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1377)
at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:178)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
... 1 more
作为检查，我打开了一个干净的 shell，而不是导入模块，我只是在交互式 shell 中定义了 str2num 函数和 UDF。然后我输入了最后一个函数的内容，并做了同样的最后检查：
df2.show()
这一次，我得到了我期待的转换后的 DataFrame。
为什么交互输入函数可以，从模块读入函数却不行？我知道它正在读取模块，因为常规函数 str2num 有效。
顺便说一句，您使用的是哪个 spark 版本？
像这样将函数转换为 UDF：
str2numUDF = F.udf(str2num, t.IntegerType())
这里不需要 lambda 函数。
我有同样的错误并跟踪堆栈跟踪。
就我而言，我正在构建一个 Egg 文件，然后通过 --py-files 选项将其传递给 spark。
关于错误，我认为这归结为这样一个事实，即当您调用 F.udf(str2num, t.IntegerType()) 时，在 Spark 运行ning 之前创建了一个 UserDefinedFunction 实例，因此它有一个空的参考一些 SparkContext，称之为 sc。当你 运行 UDF 时，sc._pickled_broadcast_vars 被引用，这会在你的输出中抛出 AttributeError。
我的解决方法是避免在 Spark 运行ning 之前创建 UDF（因此有一个活动的 SparkContext。在您的情况下，您可以更改 [=20 的定义=]
def letConvNum(df): # df is a PySpark DataFrame
#Get a list of columns that I want to transform, using the metadata Pandas DataFrame
chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist()
str2numUDF = F.udf(str2num, t.IntegerType()) # create UDF on demand
for curcol in chng_cols:
df=df.withColumn(curcol, str2numUDF(df[curcol]))
return df
注意：我还没有实际测试过上面的代码，但我自己的代码中的更改是相似的并且一切正常。
此外，对于感兴趣的 reader，请参阅 Spark code for UserDefinedFunction
如果你只在其他函数中使用 UDF，你可以这样做。
from pyspark.sql.functions import udf
class Udf(object):
def __init__(s, func, spark_type):
s.func, s.spark_type = func, spark_type
def __call__(s, *args):
return udf(s.func, s.spark_type)(*args)
myfunc_udf = Udf(myfunc, StringType())
def processing():
df_new = df.select(myfunc_udf('somefield'))
我已经为这个问题苦苦思索了整整 20 个小时。谢谢你们的解决方案！
这是我的变体，以防有人对我如何解决同样的问题感兴趣。虽然它主要来自上面的 code/responses。
这里的目的是简单地转换字符串列以显示它们的长度，但您当然可以做任何事情（我在我的主应用程序中进行数据类型检查和错误跟踪）。
我使用 udf 的目的要复杂得多，但这是我用来测试我的 udf 现在是否正常工作的方法。
假设你的数据框都是 StringType()
在我的例子中，我有 4 个字符串列
解决方案：
我制作了一个名为 myfunctions 的单独的 .py 文件
里面
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType
import logging
def str2num(text):
if type(text) == None or text == '' or text == 'NULL' or text == 'null':
return 0
else:
return len(text)
def letConvNum(df, columns):
str2numUDF = F.udf(str2num, IntegerType())
logging.info(columns)
index = 0
for curcol in columns:
df = df.withColumn(curcol, str2numUDF(df[curcol]))
index += 1
return df
然后在我的 main class
将新的 .py 文件添加到您的 sparkContext
#my understanding is that this insures your function is added to a spark across all nodes
sc.addPyFile("./myfunctions.py")
#dynamically create headers based on config -simplified for example
schemaString = "YearMonth,IMEI,IMSI,MSISDN"
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(",")]
schema = StructType(fields)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='false', inferschema='false', delimiter='|').load('/app/teacosy/invictus/kenya/SAF_QUALCOMM_IMEI_20170321.txt', schema=schema)
#read and write file to get parquet. please note this was to optimize MASSIVE files 50-200g
df.write.parquet("data.parquet", mode='overwrite')
dataframe = sqlContext.read.parquet("data.parquet")
df2 = mf.letConvNum(dataframe, schemaString.split(","))
df2.show()
输入：
+---------+---------------+---------------+------------+
|YearMonth| IMEI| IMSI| MSISDN|
+---------+---------------+---------------+------------+
| 201609|869859025975610|639021005869699|254724884336|
| 201609|359521062182040|639021025339132|254721224577|
| 201609|353121070662770|639021025339132|254721224577|
| 201609|868096015837410|639021025339132|254721224577|
| 201609|866204020015610|639021025339132|254721224577|
| 201609|356051060479107|639028040455896|254710404131|
| 201609|353071062803703|639027641207269|254725555262|
| 201609|356899067316490|639027841002602|254711955201|
| 201609|860357020164930|639028550063234|254715570856|
| 201609|862245026673900|639028940332785|254728412070|
| 201609|352441075290910|639029340152407|254714582871|
| 201609|862074027499277|639029340152407|254714582871|
| 201609|357036073532528|639028500408346|254715408346|
| 201609|356546060475230|639021011628783|254722841516|
| 201609|356546060475220|639021011628783|254722841516|
| 201609|866838023727117|639028840277749|254718492024|
| 201609|354210053950950|639029440054836|254729308302|
| 201609|866912020393040|639029870328080|254725528182|
| 201609|357921070054540|639028340694869|254710255083|
| 201609|357977056264767|639027141561199|254721977494|
输出：
+---------+----+----+------+
|YearMonth|IMEI|IMSI|MSISDN|
+---------+----+----+------+
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
我希望这能帮助那些在看到他们的 pyspark 应用程序冻结或挂起时苦苦挣扎的人……真令人沮丧，我的天啊……
我认为更简洁的解决方案是使用 udf 装饰器来定义您的 udf 函数：
from pyspark.sql.functions as F
@F.udf
def str2numUDF(text):
if type(text)==None or text=='' or text=='NULL' or text=='null':
return 0
elif len(text)==1:
return ord(text)
else:
newnum=''
for lettr in text:
newnum=newnum+str(ord(lettr))
return int(newnum)
使用此解决方案，udf 不会引用任何其他函数，因此不会向您抛出任何错误。
对于某些旧版本的 spark，装饰器不支持类型化的 udf，您可能必须按如下方式定义自定义装饰器：
from pyspark.sql.functions as F
from pyspark.sql.types as t
# Custom udf decorator which accept return type
def udf_typed(returntype=t.StringType()):
def _typed_udf_wrapper(func):
return F.udf(func, returntype)
return _typed_udf_wrapper
@udf_typed(t.IntegerType())
def my_udf(x)
return int(x)
我有一个我编写的模块，其中包含作用于 PySpark DataFrames 的函数。他们对 DataFrame 中的列进行转换，然后 return 一个新的 DataFrame。这是代码示例，已缩短为仅包含其中一个功能：
from pyspark.sql import functions as F
from pyspark.sql import types as t
import pandas as pd
import numpy as np
metadta=pd.DataFrame(pd.read_csv("metadata.csv")) # this contains metadata on my dataset
def str2num(text):
if type(text)==None or text=='' or text=='NULL' or text=='null':
return 0
elif len(text)==1:
return ord(text)
else:
newnum=''
for lettr in text:
newnum=newnum+str(ord(lettr))
return int(newnum)
str2numUDF = F.udf(lambda s: str2num(s), t.IntegerType())
def letConvNum(df): # df is a PySpark DataFrame
#Get a list of columns that I want to transform, using the metadata Pandas DataFrame
chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist()
for curcol in chng_cols:
df=df.withColumn(curcol, str2numUDF(df[curcol]))
return df
这就是我的模块，称之为 mymodule.py。如果我启动 PySpark shell，然后执行以下操作：
import mymodule as mm
myf=sqlContext.sql("select * from tablename lim 10")
我检查了 myf (PySpark DataFrame)，没问题。我通过尝试使用 str2num 函数检查我是否实际导入了 mymodule：
mm.str2num('a')
97
所以它实际上是在导入模块。那么如果我试试这个：
df2=mm.letConvNum(df)
并执行此操作以检查它是否有效：
df2.show()
它尝试执行操作，但随后崩溃：
16/03/10 16:10:44 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 365)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
16/03/10 16:10:44 ERROR TaskSetManager: Task 0 in stage 1.0 failed 1 times; aborting job
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/dataframe.py", line 256, in show
print(self._jdf.showString(n, truncate))
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__
File "/usr/hdp/2.3.4.0-3485/spark/python/pyspark/sql/utils.py", line 36, in deco
return f(*a, **kw)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o7299.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 1 times, most recent failure: Lost task 0.0 in stage 1.0 (TID 365, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage.apply(DAGScheduler.scala:1271)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage.apply(DAGScheduler.scala:1270)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1270)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed.apply(DAGScheduler.scala:697)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed.apply(DAGScheduler.scala:697)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:697)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1496)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1458)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)
at org.apache.spark.util.EventLoop$$anon.run(EventLoop.scala:48)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1824)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1837)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1850)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:215)
at org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:207)
at org.apache.spark.sql.DataFrame$$anonfun$collect.apply(DataFrame.scala:1385)
at org.apache.spark.sql.DataFrame$$anonfun$collect.apply(DataFrame.scala:1385)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
at org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:1903)
at org.apache.spark.sql.DataFrame.collect(DataFrame.scala:1384)
at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1314)
at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1377)
at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:178)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "test2.py", line 16, in <module>
str2numUDF=F.udf(lambda s: str2num(s), t.IntegerType())
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1460, in udf
return UserDefinedFunction(f, returnType)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1422, in __init__
self._judf = self._create_judf(name)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1430, in _create_judf
pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command, self)
File "/usr/hdp/2.3.4.0-3485/spark/python/lib/pyspark.zip/pyspark/rdd.py", line 2317, in _prepare_for_python_RDD
[x._jbroadcast for x in sc._pickled_broadcast_vars],
AttributeError: 'NoneType' object has no attribute '_pickled_broadcast_vars'
at org.apache.spark.api.python.PythonRunner$$anon.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:397)
at org.apache.spark.sql.execution.BatchPythonEvaluation$$anonfun$doExecute.apply(python.scala:362)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$$anonfun$apply.apply(RDD.scala:710)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
... 1 more
作为检查，我打开了一个干净的 shell，而不是导入模块，我只是在交互式 shell 中定义了 str2num 函数和 UDF。然后我输入了最后一个函数的内容，并做了同样的最后检查：
df2.show()
这一次，我得到了我期待的转换后的 DataFrame。
为什么交互输入函数可以，从模块读入函数却不行？我知道它正在读取模块，因为常规函数 str2num 有效。
顺便说一句，您使用的是哪个 spark 版本？
像这样将函数转换为 UDF：
str2numUDF = F.udf(str2num, t.IntegerType())
这里不需要 lambda 函数。
我有同样的错误并跟踪堆栈跟踪。
就我而言，我正在构建一个 Egg 文件，然后通过 --py-files 选项将其传递给 spark。
关于错误，我认为这归结为这样一个事实，即当您调用 F.udf(str2num, t.IntegerType()) 时，在 Spark 运行ning 之前创建了一个 UserDefinedFunction 实例，因此它有一个空的参考一些 SparkContext，称之为 sc。当你 运行 UDF 时，sc._pickled_broadcast_vars 被引用，这会在你的输出中抛出 AttributeError。
我的解决方法是避免在 Spark 运行ning 之前创建 UDF（因此有一个活动的 SparkContext。在您的情况下，您可以更改 [=20 的定义=]
def letConvNum(df): # df is a PySpark DataFrame
#Get a list of columns that I want to transform, using the metadata Pandas DataFrame
chng_cols=metadta[(metadta.comments=='letter conversion to num')].col_name.tolist()
str2numUDF = F.udf(str2num, t.IntegerType()) # create UDF on demand
for curcol in chng_cols:
df=df.withColumn(curcol, str2numUDF(df[curcol]))
return df
注意：我还没有实际测试过上面的代码，但我自己的代码中的更改是相似的并且一切正常。
此外，对于感兴趣的 reader，请参阅 Spark code for UserDefinedFunction
如果你只在其他函数中使用 UDF，你可以这样做。
from pyspark.sql.functions import udf
class Udf(object):
def __init__(s, func, spark_type):
s.func, s.spark_type = func, spark_type
def __call__(s, *args):
return udf(s.func, s.spark_type)(*args)
myfunc_udf = Udf(myfunc, StringType())
def processing():
df_new = df.select(myfunc_udf('somefield'))
我已经为这个问题苦苦思索了整整 20 个小时。谢谢你们的解决方案！
这是我的变体，以防有人对我如何解决同样的问题感兴趣。虽然它主要来自上面的 code/responses。
这里的目的是简单地转换字符串列以显示它们的长度，但您当然可以做任何事情（我在我的主应用程序中进行数据类型检查和错误跟踪）。
我使用 udf 的目的要复杂得多，但这是我用来测试我的 udf 现在是否正常工作的方法。
假设你的数据框都是 StringType() 在我的例子中，我有 4 个字符串列
解决方案：
我制作了一个名为 myfunctions 的单独的 .py 文件
里面
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType
import logging
def str2num(text):
if type(text) == None or text == '' or text == 'NULL' or text == 'null':
return 0
else:
return len(text)
def letConvNum(df, columns):
str2numUDF = F.udf(str2num, IntegerType())
logging.info(columns)
index = 0
for curcol in columns:
df = df.withColumn(curcol, str2numUDF(df[curcol]))
index += 1
return df
然后在我的 main class 将新的 .py 文件添加到您的 sparkContext
#my understanding is that this insures your function is added to a spark across all nodes
sc.addPyFile("./myfunctions.py")
#dynamically create headers based on config -simplified for example
schemaString = "YearMonth,IMEI,IMSI,MSISDN"
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split(",")]
schema = StructType(fields)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='false', inferschema='false', delimiter='|').load('/app/teacosy/invictus/kenya/SAF_QUALCOMM_IMEI_20170321.txt', schema=schema)
#read and write file to get parquet. please note this was to optimize MASSIVE files 50-200g
df.write.parquet("data.parquet", mode='overwrite')
dataframe = sqlContext.read.parquet("data.parquet")
df2 = mf.letConvNum(dataframe, schemaString.split(","))
df2.show()
输入：
+---------+---------------+---------------+------------+
|YearMonth| IMEI| IMSI| MSISDN|
+---------+---------------+---------------+------------+
| 201609|869859025975610|639021005869699|254724884336|
| 201609|359521062182040|639021025339132|254721224577|
| 201609|353121070662770|639021025339132|254721224577|
| 201609|868096015837410|639021025339132|254721224577|
| 201609|866204020015610|639021025339132|254721224577|
| 201609|356051060479107|639028040455896|254710404131|
| 201609|353071062803703|639027641207269|254725555262|
| 201609|356899067316490|639027841002602|254711955201|
| 201609|860357020164930|639028550063234|254715570856|
| 201609|862245026673900|639028940332785|254728412070|
| 201609|352441075290910|639029340152407|254714582871|
| 201609|862074027499277|639029340152407|254714582871|
| 201609|357036073532528|639028500408346|254715408346|
| 201609|356546060475230|639021011628783|254722841516|
| 201609|356546060475220|639021011628783|254722841516|
| 201609|866838023727117|639028840277749|254718492024|
| 201609|354210053950950|639029440054836|254729308302|
| 201609|866912020393040|639029870328080|254725528182|
| 201609|357921070054540|639028340694869|254710255083|
| 201609|357977056264767|639027141561199|254721977494|
输出：
+---------+----+----+------+
|YearMonth|IMEI|IMSI|MSISDN|
+---------+----+----+------+
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
| 6| 15| 15| 12|
我希望这能帮助那些在看到他们的 pyspark 应用程序冻结或挂起时苦苦挣扎的人……真令人沮丧，我的天啊……
我认为更简洁的解决方案是使用 udf 装饰器来定义您的 udf 函数：
from pyspark.sql.functions as F
@F.udf
def str2numUDF(text):
if type(text)==None or text=='' or text=='NULL' or text=='null':
return 0
elif len(text)==1:
return ord(text)
else:
newnum=''
for lettr in text:
newnum=newnum+str(ord(lettr))
return int(newnum)
使用此解决方案，udf 不会引用任何其他函数，因此不会向您抛出任何错误。
对于某些旧版本的 spark，装饰器不支持类型化的 udf，您可能必须按如下方式定义自定义装饰器：
from pyspark.sql.functions as F
from pyspark.sql.types as t
# Custom udf decorator which accept return type
def udf_typed(returntype=t.StringType()):
def _typed_udf_wrapper(func):
return F.udf(func, returntype)
return _typed_udf_wrapper
@udf_typed(t.IntegerType())
def my_udf(x)
return int(x)