时间戳

首页 > 解决方案 > 在 PySpark 中展开 json 列 - 架构问题 - AttributeError: 'tuple' object has no attribute 'name'

python - 在 PySpark 中展开 json 列 - 架构问题 - AttributeError: 'tuple' object has no attribute 'name'

问题描述

我正在使用 pyspark 从多行 json 对象中提取数据。我可以读取文件，但无法解析几何列的内容。

整个表格的示例如下所示。

+--------------------+--------------------+-------+
|            geometry|          properties|   type|
+--------------------+--------------------+-------+
|{[13.583336, 37.2...|{AGRIGENTO, AGRIG...|Feature|
|{[13.584538, 37.3...|{AGRIGENTO, AGRIG...|Feature|
|{[13.657838, 37.3...|{FAVARA, AGRIGENT...|Feature|
|{[13.846247, 37.3...|{CANICATTÌ, AGRI...|Feature|
|{[13.616626, 37.4...|{ARAGONA, AGRIGEN...|Feature|
|{[13.108426, 37.6...|{SAMBUCA DI SICIL...|Feature|
|{[16.709313, 41.0...|{GRUMO APPULA, BA...|Feature|
|{[12.670994, 41.4...|{NETTUNO, ROMA, 6...|Feature|
|{[12.501805, 42.1...|{CASTELNUOVO DI P...|Feature|
|{[12.608105, 41.4...|{ANZIO, ROMA, b54...|Feature|
+--------------------+--------------------+-------+

这是json几何列的单行格式

"geometry":{"type":"Point","coordinates":[13.583336,37.270182]}

当我提取模式时，这就是它的样子

StructType(List(StructField("geometry",StructType(List(StructField("coordinates",ArrayType(DoubleType,true),true),StructField("type",StringType,true))),true)

但是，当我尝试在 PySpark 中设置架构以导入数据时，出现以下错误。

AttributeError：“元组”对象没有属性“名称”

这是我正在使用的代码。

from pyspark.sql.types import StructField, StructType, StringType, FloatType, ArrayType, DoubleType
import pyspark.sql.functions as F

df = spark.read.option("multiLine", False).option("mode", "PERMISSIVE").json('Italy/it_countrywide-addresses-country.geojson')


schema = StructType([
    (StructField("coordinates",
                 ArrayType(DoubleType())),
     StructField("type",StringType()))
])

df.withColumn("geometry", F.from_json("geometry", schema)).select(col('geometry.*')).show()

我欢迎你的意见。

标签： pythonjsonapache-sparkpysparkapache-spark-sql

解决方案

最终，我的目标是读取 json 文件并访问嵌套值。我收到的错误是因为我没有正确创建架构。纠正此错误的最佳方法是避免手动创建架构。

为此，我使用了您可以通过在 json 文件上调用 .schema 来创建的架构。这解决了自己创建模式的任何问题。

这样做的缺点是您实际上是两次导入文件，毫无疑问，这可以进一步优化以避免这种情况。

json_schema = spark.read.option("multiLine", False).option("mode", "PERMISSIVE").json('Italy/it_countrywide-addresses-country.geojson').schema 
df_with_schema = spark.read.option("multiLine", False).option("mode", "PERMISSIVE").schema(json_schema).json('Italy/it_countrywide-addresses-country.geojson') 
df_with_schema.printSchema() 

# Select coordinates array
coordinates = df_with_schema.select(F.col('geometry.coordinates'))

# select single value from coordinates array
single_value_from_coordinates_array = df_with_schema.select(F.col('geometry.coordinates')[0])

# create my own dataframe choosing multiple columns from json file
multi_columns = df_with_schema.select(F.col('geometry.coordinates'), F.col('properties.city'))

推荐阅读