問題陳述：在升級 Databricks 運行時版本時，重復列在創建資料幀時拋出錯誤。在較低的運行時，資料框被創建，并且由于下游不需要重復的列，它被簡單地排除在選擇中。
檔案位置：存盤在 ADLS Gen2 (Azure) 上的 Json 檔案。集群模式：標準
代碼：我們在 Azure Databricks 中讀取它，如下所示。
intermediate_df = spark.read.option("multiline","true").json(f"{path}/IN-109418_Part_1.json")
json 檔案是嵌套的，其中一個tags是重復的列（下圖）。讀入資料框后，我們選擇所需的列。我們并不要求這個重復的tags反正。
以前，我們在 Databricks 運行時 7.3LTS(Spark3.0.1) 上運行，它創建了包含重復列的資料框，但由于我們沒有進一步使用它，因此并沒有受到影響。
但是，我們現在正在升級到運行時 9.1LTS(Spark3.1.2)，它會在創建資料幀本身時引發關于列重復的錯誤。錯誤資訊：Found duplicate column(s) in the data schema: `tags`
圖片重復列：- json 檔案中的重復列：標簽。在運行時 7.3LTS(Spark3.0.1) 中成功創建了資料框
結論：我嘗試在讀取資料幀后立即選擇列，但沒有成功。我有一種預感，因為現在 Databricks 的升級運行時版本在默認情況下更傾向于 Delta 表（增量表不支持其中的重復列），可能有一個屬性我們必須關閉以忽略此檢查整個筆記本或只是在讀入資料幀時。
雖然這個確切的錯誤發生在 json 上，但我相信如果其他檔案格式（如 csv）有重復的列，它可能會發生。
該檔案是完全嵌套的，為所有必需的列定義架構并不是很實用，因為它很乏味，并且在將來需要更多列的情況下容易出錯（這將是次要解決方案）。檔案由供應商使用自動化流程生成，預計所有檔案將保持與已交付歷史檔案相同的格式。
運行時 9.1LTS(Spark3.1.2) 上的完整錯誤：
AnalysisException Traceback (most recent call last)
<command-4270018894919110> in <module>
----> 1 intermediate_df = spark.read.option("multiline","true").json(f"{path}/IN-109418_Part_1.json")
/databricks/spark/python/pyspark/sql/readwriter.py in json(self, path, schema, primitivesAsString, prefersDecimal, allowComments, allowUnquotedFieldNames, allowSingleQuotes, allowNumericLeadingZero, allowBackslashEscapingAnyCharacter, mode, columnNameOfCorruptRecord, dateFormat, timestampFormat, multiLine, allowUnquotedControlChars, lineSep, samplingRatio, dropFieldIfAllNull, encoding, locale, pathGlobFilter, recursiveFileLookup, allowNonNumericNumbers, modifiedBefore, modifiedAfter)
370 path = [path]
371 if type(path) == list:
--> 372 return self._df(self._jreader.json(self._spark._sc._jvm.PythonUtils.toSeq(path)))
373 elif isinstance(path, RDD):
374 def func(iterator):
/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
1302
1303 answer = self.gateway_client.send_command(command)
-> 1304 return_value = get_return_value(
1305 answer, self.gateway_client, self.target_id, self.name)
1306
/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
121 # Hide where the exception came from that shows a non-Pythonic
122 # JVM exception message.
--> 123 raise converted from None
124 else:
125 raise
AnalysisException: Found duplicate column(s) in the data schema: `tags`
編輯：評論預先定義架構。
uj5u.com熱心網友回復：
請使用 json.load 將 json 轉換為字典并處理重復鍵
import json
#test json
test_json = """[
{"id": 1,
"tags": "test1",
"tags": "test1"},
{"id": 2,
"tags": "test2",
"tags": "test2",
"tags": "test3"}]
"""
#function to handle duplicate keys:
def value_resolver(pairs):
d = {}
i=1
for k, v in pairs:
if k in d:
d[k str(i)] = v
i =1
else:
d[k] = v
return d
#load
our_dict = json.loads(test_json, object_pairs_hook=value_resolver)
print(our_dict)
>> [{'id': 1, 'tags': 'test1', 'tags1': 'test1'}, {'id': 2, 'tags': 'test2', 'tags1': 'test2', 'tags2': 'test3'}]
#dict to dataframe
df = spark.createDataFrame(our_dict)
df.show()
--- ----- ----- -----
| id| tags|tags1|tags2|
--- ----- ----- -----
| 1|test1|test1| null|
| 2|test2|test2|test3|
--- ----- ----- -----
uj5u.com熱心網友回復：
目前在spark 檔案中沒有這方面的選項。對于具有重復鍵值的 jsons 的有效性以及如何處理它們，似乎也存在不同的意見/標準（SO 討論）。
提供沒有重復鍵欄位的模式會導致加載成功。它采用 json 中最后一個鍵的值。
架構取決于您的源檔案。
測驗檔案
{
"id": 1,
"tags": "test1",
"tags": "test2"
}
Python
from pyspark.sql.types import *
schema = StructType([
StructField('id', LongType(), True),
StructField('tags', StringType(), True)
])
df = spark.read.schema(schema).json("test.json", multiLine=True)
df.show()
--- -----
| id| tags|
--- -----
| 1|test2|
--- -----
在 pyspark 3.1.1 上本地運行
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/360761.html