for循环中往delta table里增量写数据并支持重跑daily数据的代码

以下是一个示例代码，用于在for循环中往Delta Table里增量写数据并支持重跑daily数据：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.appName("DeltaTableIngestion").getOrCreate()

# 定义Delta Table路径和数据源路径
deltaTablePath = "/path/to/deltaTable"
dataSourcePath = "/path/to/dataSource"

# 读取已经存在的Delta Table
deltaTable = spark.read.format("delta").load(deltaTablePath)

# 遍历数据源路径下的所有文件，并将其读取为DataFrame
for file in os.listdir(dataSourcePath):
    if file.endswith(".csv"):
        filePath = os.path.join(dataSourcePath, file)
        df = spark.read.format("csv").option("header", True).load(filePath)
        
        # 对于daily数据，我们需要删除原有的数据并重新写入新数据，这里使用overwrite模式
        # 对于增量数据，我们使用append模式
        if "daily" in file:
            deltaTable.filter(deltaTable.date != df.select("date").collect()[0][0]).write.mode("overwrite").format("delta").save(deltaTablePath)
            df.write.mode("append").format("delta").save(deltaTablePath)
        else:
            df.write.mode("append").format("delta").save(deltaTablePath)

以上代码中，我们使用了Delta Table的功能来实现增量写入和重跑。对于daily数据，我们首先找到该数据对应的日期，然后使用Delta Table的过滤功能来删除原有日期的数据，最后重新写入新数据。对于增量数据，我们直接使用append模式来将其写入Delta Table。注意，我们需要在每次循环结束后关闭SparkSession。