pyspark： TypeError： IntegerType 不能接受类型<类型'unicode'中的对象>

Question

在Spark集群上使用pyspark编程数据非常大，而且是碎片状的，因此无法加载到内存中，也无法轻松检查数据的完整性

基本上就是

af.b Current%20events 1 996
af.b Kategorie:Musiek 1 4468
af.b Spesiaal:RecentChangesLinked/Gebruikerbespreking:Freakazoid 1 5209
af.b Spesiaal:RecentChangesLinked/Sir_Arthur_Conan_Doyle 1 5214

维基百科数据:

我从aws S3中读取它，然后尝试在pyspark解释器中使用以下python代码构建spark Dataframe:

parts = data.map(lambda l: l.split())
wikis = parts.map(lambda p: (p[0], p[1],p[2],p[3]))

fields = [StructField("project", StringType(), True),
StructField("title", StringType(), True),
StructField("count", IntegerType(), True),
StructField("byte_size", StringType(), True)] 
schema = StructType(fields) 
df = sqlContext.createDataFrame(wikis, schema)

一切看起来都很好，只有createDataFrame给了我错误

Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/spark/python/pyspark/sql/context.py", line 404, in   createDataFrame
rdd, schema = self._createFromRDD(data, schema, samplingRatio)
File "/usr/lib/spark/python/pyspark/sql/context.py", line 298, in _createFromRDD
_verify_type(row, schema)
File "/usr/lib/spark/python/pyspark/sql/types.py", line 1152, in _verify_type
_verify_type(v, f.dataType)
File "/usr/lib/spark/python/pyspark/sql/types.py", line 1136, in _verify_type
raise TypeError("%s can not accept object in type %s" % (dataType, type(obj)))
TypeError: IntegerType can not accept object in type <type 'unicode'>

为什么我不能设置第三列，应该计数为IntegerType ?如何解决这个问题?

Answer 1

正如cheneson所指出的，你传递了错误的类型。

假设你的data是这样的:

data = sc.parallelize(["af.b Current%20events 1 996"])

在第一张地图之后你得到RDD[List[String]]:

parts = data.map(lambda l: l.split())
parts.first()
## ['af.b', 'Current%20events', '1', '996']

第二个映射将其转换为元组(String, String, String, String):

wikis = parts.map(lambda p: (p[0], p[1], p[2],p[3]))
wikis.first()
## ('af.b', 'Current%20events', '1', '996')

您的schema声明第三列是整数:

[f.dataType for f in schema.fields]
## [StringType, StringType, IntegerType, StringType]

Schema主要用于避免全表扫描来推断类型，并且不执行任何类型转换。

您可以在上次映射时强制转换您的数据:

wikis = parts.map(lambda p: (p[0], p[1], int(p[2]), p[3]))

或者将count定义为StringType并强制转换列

fields[2] = StructField("count", StringType(), True)
schema = StructType(fields) 
wikis.toDF(schema).withColumn("cnt", col("count").cast("integer")).drop("count")

count是SQL中的保留字，不应该用作列名。在Spark中，它会在某些上下文中正常工作，而在另一些上下文中失败。

Answer 2

使用apache 2.0，您可以让spark推断数据的模式。总的来说，您需要在解析器函数中强制转换，如上所述:

"当schema为None时，它将尝试从数据中推断模式(列名和类型)，该数据应该是Row, namedtuple或dict的RDD。"