我正在尝试定义我的udf，以清除标记中的html文本。以下代码可以正常工作：
from bs4 import BeautifulSoup
from pyspark.sql.functions import udf
text = '<p>Tervetuloa leikkimään, laulamaan, loruilemaan, liikkumaan, taiteilemaan ja tutkimaan leikkipuiston<br>perheaamuun! Leikki- ja toimintaympäristö mahdollistavat vanhemman ja lapsen yhteisen puuhan ja leikin<br>ja lapset saavat leikkiseuraa.<br>Vanhemmilla on mahdollisuus tutustua muihin lapsiperheisiin ja lapset saavat leikkiseuraa. Vanhemmat ja<br>lapset voivat osallistua toiminnan suunnittel'
text_clr = BeautifulSoup(text, 'html.parser').get_text()
print(text_clr)
结果正确：
Tervetuloa leikkimään, laulamaan, loruilemaan, liikkumaan, taiteilemaan ja tutkimaan leikkipuistonperheaamuun! Leikki- ja toimintaympäristö mahdollistavat vanhemman ja lapsen yhteisen puuhan ja leikinja lapset saavat leikkiseuraa.Vanhemmilla on mahdollisuus tutustua muihin lapsiperheisiin ja lapset saavat leikkiseuraa. Vanhemmat jalapset voivat osallistua toiminnan suunnittel
然后我定义我的udf：
from bs4 import BeautifulSoup
from pyspark.sql.functions import udf
spark.udf.register("soup_udf",
lambda text_clr: BeautifulSoup(text, 'html.parser').get_text() if not text is None else 'NA',
"string")
text1 = '<p>Tervetuloa leikkimään, laulamaan, loruilemaan, liikkumaan, taiteilemaan ja tutkimaan leikkipuiston<br>perheaamuun! Leikki- ja toimintaympäristö mahdollistavat vanhemman ja lapsen yhteisen puuhan ja leikin<br>ja lapset saavat leikkiseuraa.<br>Vanhemmilla on mahdollisuus tutustua muihin lapsiperheisiin ja lapset saavat leikkiseuraa. Vanhemmat ja<br>lapset voivat osallistua toiminnan suunnittel'
text_clr1 = soup_udf(text1)
print(text_clr1)
结果为：Column<b'<lambda>(<p>Tervetuloa leikkim\xc3\xa4\xc3\xa4n, laulamaan, loruilemaan, liikkumaan, taiteilemaan ja tutkimaan leikkipuiston<br>perheaamuun! Leikki- ja toimintaymp\xc3\xa4rist\xc3\xb6 mahdollistavat vanhemman ja lapsen yhteisen puuhan ja leikin<br>ja lapset saavat leikkiseuraa.<br>Vanhemmilla on mahdollisuus tutustua muihin lapsiperheisiin ja lapset saavat leikkiseuraa. Vanhemmat ja<br>lapset voivat osallistua toiminnan suunnittel)'>
为什么结果不同？为什么它不能像udf一样工作？
问题的第二部分是，我想在数据框中使用我的soup_udf。
display(dfAll4.select("id", soup_udf("desc").alias("desc_clr")).distinct())
dfAll4.select("id", soup_udf("desc").alias("desc_clr")).distinct().show(10,truncate=200)
因此，我收到一条冗长的错误消息，我不明白：-（
DataFrame[id: string, desc_clr: string]
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-118-aa5fcd68d914> in <module>
20 #display(df.select("id", squared_udf("id").alias("id_squared")))
21 display(dfAll4.select("id", soup_udf("desc").alias("desc_clr")).distinct())
---> 22 dfAll4.select("id", soup_udf("desc").alias("desc_clr")).distinct().show(10,truncate=200)
23 #dfAll4.withColumn("desc_clr", soup_udf(dfAll4.desc)).select("desc_clr").distinct().show(10, truncate=200)
24 #dfAll4.select("desc", soup_udf(dfAll4.desc).alias("desc_clr")).distinct().show(10, truncate=200)
/usr/lib/spark-2.4.4/python/pyspark/sql/dataframe.py in show(self, n, truncate, vertical)
380 print(self._jdf.showString(n, 20, vertical))
381 else:
--> 382 print(self._jdf.showString(n, int(truncate), vertical))
383
384 def __repr__(self):
/usr/lib/spark-2.4.4/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/usr/lib/spark-2.4.4/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/usr/lib/spark-2.4.4/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling o3334.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 426.0 failed 1 times, most recent failure: Lost task 2.0 in stage 426.0 (TID 28703, localhost, executor driver): net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict (for bs4.element.NavigableString)
at net.razorvine.pickle.objects.ClassDictConstructor.construct(ClassDictConstructor.java:23)
at net.razorvine.pickle.Unpickler.load_reduce(Unpickler.java:707)
at net.razorvine.pickle.Unpickler.load_newobj(Unpickler.java:711)
at net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:259)
at net.razorvine.pickle.Unpickler.load(Unpickler.java:99)
at net.razorvine.pickle.Unpickler.loads(Unpickler.java:112)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:90)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:89)
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage16.agg_doAggregateWithKeys_0$(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage16.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
at org.apache.spark.scheduler.Task.run(Task.scala:123)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1889)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1877)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1876)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1876)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:926)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:926)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2110)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2059)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2048)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:737)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:365)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3389)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2550)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2550)
at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3370)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3369)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2550)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2764)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:254)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:291)
at sun.reflect.GeneratedMethodAccessor81.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
Caused by: net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict (for bs4.element.NavigableString)
at net.razorvine.pickle.objects.ClassDictConstructor.construct(ClassDictConstructor.java:23)
at net.razorvine.pickle.Unpickler.load_reduce(Unpickler.java:707)
at net.razorvine.pickle.Unpickler.load_newobj(Unpickler.java:711)
at net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:259)
at net.razorvine.pickle.Unpickler.load(Unpickler.java:99)
at net.razorvine.pickle.Unpickler.loads(Unpickler.java:112)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:90)
at org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:89)
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage16.agg_doAggregateWithKeys_0$(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage16.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
at org.apache.spark.scheduler.Task.run(Task.scala:123)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
有人可以帮忙吗？
非常感谢，艾丽西亚
也许是因为buautifulsoup依赖太多。试试这个库。它不依赖于其他库