以下的細節要求:
1。我使用磚筆記本從卡夫卡的主題和寫作讀取數據到ADLS Gen2容器即。我著陸層。
2。我用火花從卡夫卡和寫代碼來讀取數據到著陸層。
3所示。下一步是,我正在讀Json文件從著陸層搬到銅層,這是另一個容器ADLS代。為此,我使用自動裝卸機使用自動裝卸機與達美住表創建表。
這是相同的代碼:
@dlt.table (
name =表名,
評論= "創建青銅表”,
table_properties = {
“質量”:“青銅”
}
)
def Bronze_Table_Create ():
回報(
火花
.readStream
.format (“cloudFiles”)
.option (“cloudFiles。格式”、“json”)
.option (“cloudFiles。schemaLocation”, schemaLocation)
.option (“cloudFiles。在ferColumnTypes", "true")
.option (“cloudFiles。schemaEvolutionMode”、“救援”)
.load (sourcelocation)
)
4所示。這段代碼對我來說很不錯,它推斷模式。然而,我有一個場景中,我嚐試處理。我一步一步提到如下:
我。我想驗證模式,所以如果有任何的變化模式,我應該得到通知和工作應該失敗。我可以處理它SchemaEvolutionMode。但是,我的情況是完全不同的。在我的情況下,我在一列RawData,這是類型的對象,並沒有指定的模式。它將動態值,因為如果我推斷模式和應用模式驗證,然後每次將帶來新的模式和模式失配誤差。
二世。是有任何解決方案,我可以排除嗎RawData列從模式驗證,所以我允許這個專欄有任何類型的數據。
我掙紮著從很長一段時間。任何幫助是有益的。請讓我知道如果需要任何額外的細節。
示例Json:
{
"頭":{
“SchemaVersion”:“1.0”,
“EventId”:“123”,
“EventTime_UTC”:“2022-09-22 16:18:16”,
“環境”:“開發”,
},
“有效載荷”:{
" RawData ": {
“CusID”:“12345”,
“狀態”:“等待”,
“LastModifiedAt”:“2022-09-22 16:18:12”,
:“ContainerName違約,
“CreatedAt”:“2022-09-22 16:18:11”* RawData中的數據不一致喜歡它可以有不同的列*
},
"數據":{
“CustID”:“12345”,
“ArrayKeys”:(
{
:“ArrayName WorkHistory”,
:“ArrayKeyName SampleId”
}
]
}
}
}
