Spark Sql: TypeError("StructType 不能接受类型为 %s 的对象" % type(obj))

Posted 2023-04-17

【中文标题】Spark Sql: TypeError("StructType 不能接受类型为 %s 的对象" % type(obj))

【英文标题】：Spark Sql: TypeError("StructType can not accept object in type %s" % type(obj))

【发布时间】：2016-04-17 12:39:19

【问题描述】：

我目前正在使用 PyODBC 从 SQL Server 中提取数据，并尝试以近实时 (NRT) 方式插入 Hive 中的表。

我从源代码中获取了一行并转换为 List[Strings] 并以编程方式创建架构，但在创建 DataFrame 时，Spark 抛出 StructType 错误。

>>> cnxn = pyodbc.connect(con_string)
>>> aj = cnxn.cursor()
>>>
>>> aj.execute("select * from tjob")
<pyodbc.Cursor object at 0x257b2d0>

>>> row = aj.fetchone()

>>> row
(1127, u'', u'8196660', u'', u'', 0, u'', u'', None, 35, None, 0, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, u'', 0, None, None)
>>> rowstr = map(str,row)
>>> rowstr
['1127', '', '8196660', '', '', '0', '', '', 'None', '35', 'None', '0', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', '', '0', 'None', 'None']

>>> schemaString = " ".join([row.column_name for row in aj.columns(table='tjob')])

>>> schemaString
u'ID ExternalID Name Description Notes Type Lot SubLot ParentJobID ProductID PlannedStartDateTime PlannedDurationSeconds Capture01 Capture02 Capture03 Capture04 Capture05 Capture06 Capture07 Capture08 Capture09 Capture10 Capture11 Capture12 Capture13 Capture14 Capture15 Capture16 Capture17 Capture18 Capture19 Capture20 User UserState ModifiedDateTime UploadedDateTime'

>>> fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
>>> schema = StructType(fields)

>>> [f.dataType for f in schema.fields]
[StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType, StringType]

>>> myrdd = sc.parallelize(rowstr)

>>> myrdd.collect()
['1127', '', '8196660', '', '', '0', '', '', 'None', '35', 'None', '0', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', 'None', '', '0', 'None', 'None']

>>> schemaPeople = sqlContext.createDataFrame(myrdd, schema)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/apps/opt/cloudera/parcels/CDH-5.5.2-1.cdh5.5.2.p0.4/lib/spark/python/pyspark/sql/context.py", line 404, in createDataFrame
    rdd, schema = self._createFromRDD(data, schema, samplingRatio)
  File "/apps/opt/cloudera/parcels/CDH-5.5.2-1.cdh5.5.2.p0.4/lib/spark/python/pyspark/sql/context.py", line 298, in _createFromRDD
    _verify_type(row, schema)
  File "/apps/opt/cloudera/parcels/CDH-5.5.2-1.cdh5.5.2.p0.4/lib/spark/python/pyspark/sql/types.py", line 1132, in _verify_type
    raise TypeError("StructType can not accept object in type %s" % type(obj))
TypeError: StructType can not accept object in type <type 'str'>

【参考方案1】：

这里是错误信息的原因：

>>> rowstr
['1127', '', '8196660', '', '', '0', '', '', 'None' ... ]   
#rowstr is a list of str

>>> myrdd = sc.parallelize(rowstr)
#myrdd is a rdd of str

>>> schema = StructType(fields)
#schema is StructType([StringType, StringType, ....])

>>> schemaPeople = sqlContext.createDataFrame(myrdd, schema)
#myrdd should have been RDD([StringType, StringType,...]) but is RDD(str)

要解决这个问题，请制作正确类型的 RDD：

>>> myrdd = sc.parallelize([rowstr])

【讨论】：

至于“正确的类型”是什么？ str 的列表？ sc.parallelize()?

这是一个非常有用的答案。我花了一个多小时陷入误导性错误 TypeError: StructType can not accept object 123 in type 。感谢这篇文章，我了解到我需要使用表达式 spark.createDataFrame([(123,)], my_schema) 而不是 spark.createDataFrame([123], my_schema)

创建数据框

@adelinor 你知道为什么 spark.createDataFrame([(123,)], my_schema) 中的 123 后面要加逗号吗？

@hipokito ：逗号对于推断表达式的类型似乎很重要。

@adelinor ：逗号在处理单例（单元素元组）时尤为重要；创建单元素元组需要尾随逗号。

【参考方案2】：

我现在收到了类似的错误！

TypeError: StructType 不能接受 <class 'str'> 类型中的对象“_id”

这就是我解决它的方法。

我正在使用大量嵌套的 json 文件进行调度，json 文件由列表等字典列表组成。

例如['1127', time: '_id', '8196660', '', '', '0', '', '', 'None' ..., busstops: _id, name ]

对我来说，_id 在其他字典中重复了很多次，我通过指定字典键来解决它。

kl= spark.createDataFrame(obj_day, schema=test()) #: I get the error

但我用

解决了

kl= spark.createDataFrame(obj_day["busstops"], schema=test())