在 Pyspark 中从 Rest Api 创建数据框时出错

Posted 2023-04-15

【中文标题】在 Pyspark 中从 Rest Api 创建数据框时出错

【英文标题】：Error while creating data frame from Rest Api in Pyspark

【发布时间】：2020-07-19 06:38:10

【问题描述】：

我有一个下面的 pyspark 代码。我正在从 Rest API 读取 json 数据并尝试使用 pyspark 加载。 但我无法在 Spark 中读取 DataFrame 中的数据。有人可以帮我吗？

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

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

rdd = sc.parallelize(data)
df = spark.createDataFrame(rdd,schema)
df.show()```

**The Error message is TypeError: StructType can not accept object '[' in type <class 'str'>**
** I have been able to do using dataset in scala but i am not able to understand why its not possible using python **

导入 spark.implicits._

// 从 New York City Taxi data REST API 加载 2016 Green Taxi Trip Data 的数据 val url="https://data.cityofnewyork.us/resource/pqfs-mqru.json" val 结果 = scala.io.Source.fromURL(url).mkString

// 从 JSON 数据创建一个数据框 val taxDF = spark.read.json(Seq(result).toDS)

// 显示包含行程数据的数据框 taxDF.show()

【参考方案1】：

只为别人.. 这是对我有用的代码。Request .get 返回一个列表

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

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'
r = requests.get(url)
data_json = r.json()
df = spark.createDataFrame(data_json,schema)
display(df)