PySpark从REST API创建DataFrame报错，Scala可实现求Python解决方案

我来帮你捋清楚问题出在哪，以及怎么修复这个PySpark代码～

错误原因 #

你遇到的TypeError: StructType can not accept object '[' in type <class 'str'>，核心问题出在这一行：

rdd = sc.parallelize(data)

data是你从API读取到的完整JSON数组字符串（比如[{"a":1}, {"b":2}]这种格式），而Python的parallelize方法会把字符串当成可迭代对象，把每个字符拆成RDD的一个元素。也就是说你的RDD里现在全是单个字符（'[', '{', "a", ':'...），当你用createDataFrame(rdd, schema)时，Spark会尝试把每个字符映射到你定义的StructType里，这显然完全不匹配，所以报错了。

而你的Scala代码能成功，是因为Seq(result).toDS把整个JSON字符串作为单个元素放进了Dataset里，spark.read.json可以识别这个单个元素里的完整JSON数组，自动解析成DataFrame。

解决方案 #

这里有两种简单的修复方式，都能达到和Scala代码一样的效果：

方法1：直接用spark.read.json处理字符串（推荐） #

和Scala的思路一致，把JSON字符串包装成一个单元素的列表，然后让Spark的JSON reader来解析，还可以指定你定义好的schema来提升性能：

import urllib
from pyspark.sql.types import StructType,StructField,StringType

# 定义schema（和你原来的一样）
schema = StructType([StructField('dropoff_latitude',StringType(),True),
    StructField('dropoff_longitude',StringType(),True),
    StructField('extra',StringType(),True),
    StructField('fare_amount',StringType(),True),
    StructField('improvement_surcharge',StringType(),True),
    StructField('lpep_dropoff_datetime',StringType(),True),
    StructField('mta_tax',StringType(),True),
    StructField('passenger_count',StringType(),True),
    StructField('payment_type',StringType(),True),
    StructField('pickup_latitude',StringType(),True),
    StructField('ratecodeid',StringType(),True),
    StructField('tip_amount',StringType(),True),
    StructField('tolls_amount',StringType(),True),
    StructField('total_amount',StringType(),True),
    StructField('trip_distance',StringType(),True),
    StructField('trip_type',StringType(),True),
    StructField('vendorid',StringType(),True)
])

url = 'https://data.cityofnewyork.us/resource/pqfs-mqru.json'
data = urllib.request.urlopen(url).read().decode('utf-8')

# 把字符串包装成单元素列表，交给spark.read.json解析
df = spark.read.json([data], schema=schema)
df.show()

方法2：手动解析JSON后并行化 #

如果你想手动处理RDD，可以先把JSON字符串解析成Python字典的列表，再并行化这个列表，这样RDD的每个元素都是符合schema的字典：

import urllib
import json
from pyspark.sql.types import StructType,StructField,StringType

schema = StructType([StructField('dropoff_latitude',StringType(),True),
    # 省略其他字段，和你原来的schema一致
    StructField('vendorid',StringType(),True)
])

url = 'https://data.cityofnewyork.us/resource/pqfs-mqru.json'
data = urllib.request.urlopen(url).read().decode('utf-8')

# 把JSON字符串解析成Python字典列表
json_data = json.loads(data)
# 并行化字典列表
rdd = sc.parallelize(json_data)
# 创建DataFrame
df = spark.createDataFrame(rdd, schema)
df.show()

Scala vs Python的关键差异 #

Scala里的Seq(result).toDS是将整个JSON数组字符串作为一个元素存入Dataset，Spark的JSON reader会自动识别这个元素是完整的JSON数组并解析。而Python中直接parallelize(data)会把字符串拆成单个字符，这是两种语言处理可迭代对象的差异导致的，也是你代码失败的核心原因。

内容的提问来源于stack exchange，提问作者Pyspark Developer