這就跟你問聲好!
為什麼田野discovery_time、commit_time archive_time cloud_files_state空嗎?
我需要配置什麼當創建我的汽車裝載機嗎?
df = spark.readStream.format (cloudFiles) \ .option (“cloudFiles。格式”、“json”) \ .option (“cloudFiles。tenantId”, tenantId) \ .option (“cloudFiles。clientId”, clientId) \ .option (“cloudFiles。clientSecret”, clientSecret) \ .option (“cloudFiles。resourceGroup”, resourceGroup) \ .option (“cloudFiles。subscriptionId”, subscriptionId) \ .option (“cloudFiles。useNotifications”、“真實”)\ .option (“cloudFiles。在cludeExistingFiles", "true") \ .option("cloudFiles.schemaLocation", checkpoint_path) \ .option("cloudFiles.schemaEvolutionMode", "rescue") \ .option("recursiveFileLookup", "true") \ .option("badRecordsPath", bad_records_path) .option("multiLine", "true") .schema(dfSchema.schema) \ .load(sourceDir) #Transforming dataframe stream... df6.writeStream \ .format("delta") \ .foreachBatch(upsertToDelta) \ .option("checkpointLocation", checkpoint_path) \ .outputMode("update") \ .start(targetDir) #target folder