我是 pyspark 的新手。我希望能夠從我的 kafka 主題中讀取值。為此，我為主題中的訊息創建了一個架構。
這是我的 kafka 主題中的示例訊息：
{
"action": "string",
"id": "string",
"epoch": "long",
"entity": {
"type": "string",
"sources": [{
"items": [
{"identifier": "string"},
{"identifier": "string"},
]
}]
}
}
這是我的架構：
root
|-- value: struct (nullable = true)
| |-- action: string (nullable = true)
| |-- id: string (nullable = true)
| |-- epoch: long (nullable = true)
| |-- entity: struct (nullable = true)
| | |-- type: string (nullable = true)
| |-- entity: struct (nullable = true)
| | |-- sources: array (nullable = true)
| | | |-- element: struct (containsNull = true)
| | | | |-- items: array (nullable = true)
| | | | | |-- element: struct (containsNull = true)
| | | | | | |-- identifier: string (nullable = true)
但是，在嘗試選擇我感興趣的欄位時出現錯誤。
value_df = df.select(
from_json(col("value").cast("string"), Schema).alias("value"),
)
explode_df = value_df.select(
"value.id",
"value.action",
"value.epoch",
"value.entity.type",
"value.entity.sources.items.identifier",
)
explode_df.printSchema()
這是我得到的錯誤：
An error was encountered:
Ambiguous reference to fields StructField(entity,StructType(StructField(type,StringType,true)),true), StructField(entity,StructType(StructField(sources,ArrayType(StructType(StructField(items,ArrayType(StructType(StructField(identifier,StringType,true)),true),true)),true),true)),true)
Traceback (most recent call last):
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py", line 1671, in select
jdf = self._jdf.select(self._jcols(*cols))
File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1305, in __call__
answer, self.gateway_client, self.target_id, self.name)
File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 117, in deco
raise converted from None
pyspark.sql.utils.AnalysisException: Ambiguous reference to fields StructField(entity,StructType(StructField(type,StringType,true)),true), StructField(entity,StructType(StructField(sources,ArrayType(StructType(StructField(items,ArrayType(StructType(StructField(identifier,StringType,true)),true),true)),true),true)),true)
我相信這是因為sources和items都是陣列。有沒有辦法使用pyspark將陣列讀入df？
uj5u.com熱心網友回復：
看起來entity您的架構中有兩次，這是模棱兩可的參考。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/332713.html
標籤：阿帕奇火花 火花 apache-spark-sql 火花流 亚马逊-emr