pyspark：TypeError：整型无法接受类型中的对象

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 intepreter中使用以下python代码构建spark数据框架：

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'>

为什么我不能将第三列count设置为IntegerType？我如何解决这个问题？

Answer 1

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

当schema为None时，它将尝试从数据推断架构（列名和类型），数据应为Row、namedtuple或dict的RDD

Answer 2

正如ccheneson所指出的，您传递了错误的类型。

假设您的数据如下所示：

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')

您的模式指出，第三列是一个整数：

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

模式最常用于避免全表扫描来推断类型，并且不执行任何类型转换。

您可以在上次映射期间强制转换数据：

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

或将计数定义为StringType和cast列

fields[2] = StructField("count", StringType(), True)
schema = StructType(fields) 

wikis.toDF(schema).withColumn("cnt", col("count").cast("integer")).drop("count")

另一方面，count是SQL中的保留字，不应用作列名。在Spark中，它在某些情况下会按预期工作，而在另一种情况下会失败。