我有一个spark数据帧。我在数据帧的每一行使用一个映射函数来使用HTML解析库解析文本列，然后将解析的HTML和另外两列一起保存为新的RDD。在

最后，我希望将RDD保存为新的Spark Dataframe。这是相同的代码。在def htmlParsing(x):

""" This function takes the input text and cleans the HTML tags from it

"""

from bs4 import BeautifulSoup

row=x.asDict()

textcleaned=''

souptext=BeautifulSoup(row['desc'])

#souptext=BeautifulSoup(text)

p_tags=souptext.find_all('p')

for p in p_tags:

if p.string:

textcleaned+=p.string

ret_list= (int(row['id']),row['title'],textcleaned)

return ret_list

ret_list=sdf_rss.map(htmlParsing)

sdf_cleaned=sqlContext.createDataFrame(ret_list,['id','title','desc'])

sdf_cleaned.count()

当我回来的时候_列表。take(2) 我确实得到了正确的映射结果。同样的道理也适用于自卫队_清洁。显示()等

当我得到正确的RDD时，映射函数工作得很好。请参见下面的映射函数返回RDD的结果。在

^{pr2}$

然而，当我真的指望这两个，它抛出错误。在ret_list.count()/Users/i854319/spark/python/pyspark/rdd.pyc in count(self)

1002 3

1003 """

-> 1004 return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()

1005

1006 def stats(self):

/Users/i854319/spark/python/pyspark/rdd.pyc in sum(self)

993 6.0

994 """

--> 995 return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)

996

997 def count(self):

/Users/i854319/spark/python/pyspark/rdd.pyc in fold(self, zeroValue, op)

867 # zeroValue provided to each partition is unique from the one provided

868 # to the final reduce call

--> 869 vals = self.mapPartitions(func).collect()

870 return reduce(op, vals, zeroValue)

871

/Users/i854319/spark/python/pyspark/rdd.pyc in collect(self)

769 """

770 with SCCallSiteSync(self.context) as css:

--> 771 port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())

772 return list(_load_from_socket(port, self._jrdd_deserializer))

773

/Users/i854319/spark/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py in __call__(self, *args)

811 answer = self.gateway_client.send_command(command)

812 return_value = get_return_value(

--> 813 answer, self.gateway_client, self.target_id, self.name)

814

815 for temp_arg in temp_args:

/Users/i854319/spark/python/pyspark/sql/utils.pyc in deco(*a, **kw)

43 def deco(*a, **kw):

44 try:

---> 45 return f(*a, **kw)

46 except py4j.protocol.Py4JJavaError as e:

47 s = e.java_exception.toString()

/Users/i854319/spark/python/lib/py4j-0.9-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)

306 raise Py4JJavaError(

307 "An error occurred while calling {0}{1}{2}.\n".

--> 308 format(target_id, ".", name), value)

309 else:

310 raise Py4JError(

Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.

: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 129.0 failed 1 times, most recent failure: Lost task 2.0 in stage 129.0 (TID 189, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):

File "/Users/i854319/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main

process()

File "/Users/i854319/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process

serializer.dump_stream(func(split_index, iterator), outfile)

File "/Users/i854319/spark/python/pyspark/rdd.py", line 2346, in pipeline_func

return func(split, prev_func(split, iterator))

File "/Users/i854319/spark/python/pyspark/rdd.py", line 2346, in pipeline_func

return func(split, prev_func(split, iterator))

File "/Users/i854319/spark/python/pyspark/rdd.py", line 2346, in pipeline_func

return func(split, prev_func(split, iterator))

File "/Users/i854319/spark/python/pyspark/rdd.py", line 317, in func

return f(iterator)

File "/Users/i854319/spark/python/pyspark/rdd.py", line 1004, in

return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()

File "/Users/i854319/spark/python/pyspark/rdd.py", line 1004, in

return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()

File "", line 10, in htmlParsing

File "/Users/i854319/anaconda/lib/python2.7/site-packages/bs4/__init__.py", line 176, in __init__

elif len(markup) <= 256:

TypeError: object of type 'NoneType' has no len()