# 大批量数据插入，要么全部成功，要么全部失败
large_df.write.format("delta").mode("append").save("/path/to/table")

-- 查询特定版本
SELECT * FROM my_table VERSION AS OF 5

-- 查询特定时间点
SELECT * FROM my_table TIMESTAMP AS OF '2023-01-01 00:00:00'

-- 查看版本历史
DESCRIBE HISTORY my_table

# 自动添加新列
new_df.write.format("delta").option("mergeSchema", "true").mode("append").save("/path/to/table")

-- 手动删除列
ALTER TABLE my_table DROP COLUMN old_column

-- 重命名列
ALTER TABLE my_table RENAME COLUMN old_name TO new_name

-- 兼容类型转换
ALTER TABLE my_table ALTER COLUMN age TYPE BIGINT

# 添加检查约束
spark.sql("""
    ALTER TABLE my_table 
    ADD CONSTRAINT age_check CHECK (age >= 0 AND age <= 150)
""")

# 添加非空约束
spark.sql("""
    ALTER TABLE my_table 
    ALTER COLUMN email SET NOT NULL
""")

# 批处理写入
batch_df.write.format("delta").save("/path/to/table")

# 流式写入（相同的表）
stream_df.writeStream.format("delta").start("/path/to/table")

# 批处理读取
batch_read = spark.read.format("delta").load("/path/to/table")

# 流式读取（相同的表）
stream_read = spark.readStream.format("delta").load("/path/to/table")

# 启用自动压缩
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# 启用优化写入
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# 手动触发 Z-Order 优化
deltaTable.optimize().executeZOrderBy("department", "age")

<dependency>
    <groupId>io.delta</groupId>
    <artifactId>delta-core_2.12</artifactId>
    <version>2.4.0</version>
</dependency>

<dependency>
    <groupId>io.delta</groupId>
    <artifactId>delta-storage</artifactId>
    <version>2.4.0</version>
</dependency>

libraryDependencies += "io.delta" %% "delta-core" % "2.4.0"
libraryDependencies += "io.delta" %% "delta-storage" % "2.4.0"

# 安装 Delta Lake Python 包
pip install delta-spark

# 或者使用 conda
conda install -c conda-forge delta-spark

spark-shell --packages io.delta:delta-core_2.12:2.4.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

pyspark --packages io.delta:delta-core_2.12:2.4.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

spark-submit --packages io.delta:delta-core_2.12:2.4.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
  your_application.py

import org.apache.spark.sql.SparkSession
import io.delta.sql._

val spark = SparkSession.builder()
  .appName("DeltaLakeExample")
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
  .getOrCreate()

from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

builder = SparkSession.builder.appName("DeltaLakeExample") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

# 启用自动优化
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# 配置文件大小
spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728")  # 128MB
spark.conf.set("spark.sql.files.openCostInBytes", "4194304")     # 4MB

# 配置并行度
spark.conf.set("spark.sql.shuffle.partitions", "200")
spark.conf.set("spark.default.parallelism", "100")

# 冲突重试配置
spark.conf.set("spark.databricks.delta.retryWriteConflict.enabled", "true")
spark.conf.set("spark.databricks.delta.retryWriteConflict.limit", "3")

# 乐观锁配置
spark.conf.set("spark.databricks.delta.merge.repartitionBeforeWrite.enabled", "true")

# S3 访问配置
spark.conf.set("spark.hadoop.fs.s3a.access.key", "your-access-key")
spark.conf.set("spark.hadoop.fs.s3a.secret.key", "your-secret-key")
spark.conf.set("spark.hadoop.fs.s3a.endpoint", "s3.amazonaws.com")

# S3 性能优化
spark.conf.set("spark.hadoop.fs.s3a.multipart.size", "104857600")  # 100MB
spark.conf.set("spark.hadoop.fs.s3a.multipart.threshold", "2147483647")  # 2GB
spark.conf.set("spark.hadoop.fs.s3a.fast.upload", "true")

# HDFS 配置
spark.conf.set("spark.hadoop.dfs.client.use.datanode.hostname", "true")
spark.conf.set("spark.hadoop.dfs.client.cache.drop.behind.reads", "true")

# Azure 配置
spark.conf.set("spark.hadoop.fs.azure.account.key.your-account.dfs.core.windows.net", "your-key")
spark.conf.set("spark.hadoop.fs.azure.account.auth.type.your-account.dfs.core.windows.net", "SharedKey")

# Spark 配置
spark.sql.extensions io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog org.apache.spark.sql.delta.catalog.DeltaCatalog

# 自动优化
spark.databricks.delta.optimizeWrite.enabled true
spark.databricks.delta.autoCompact.enabled true

[
  {
    "classification": "spark-defaults",
    "properties": {
      "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
      "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
      "spark.databricks.delta.optimizeWrite.enabled": "true",
      "spark.databricks.delta.autoCompact.enabled": "true"
    }
  }
]

apiVersion: v1
kind: ConfigMap
metadata:
  name: spark-config
data:
  spark-defaults.conf: |
    spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension
    spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
    spark.databricks.delta.optimizeWrite.enabled=true
    spark.databricks.delta.autoCompact.enabled=true

from pyspark.sql import SparkSession
from delta.tables import DeltaTable
import delta

# 创建测试数据
data = [(1, "Alice", 25), (2, "Bob", 30), (3, "Charlie", 35)]
columns = ["id", "name", "age"]
df = spark.createDataFrame(data, columns)

# 写入 Delta 表
df.write.format("delta").save("/tmp/test-delta-table")

# 读取验证
delta_df = spark.read.format("delta").load("/tmp/test-delta-table")
delta_df.show()

# 检查版本
DeltaTable.forPath(spark, "/tmp/test-delta-table").history().show()

print("Delta Lake 安装验证成功！")

# 测试时间旅行
spark.sql("SELECT * FROM delta.`/tmp/test-delta-table` VERSION AS OF 0").show()

# 测试 Schema 演进
new_data = [(4, "David", 28, "Engineering")]
new_columns = ["id", "name", "age", "department"]
new_df = spark.createDataFrame(new_data, new_columns)

new_df.write.format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .save("/tmp/test-delta-table")

# 验证新 Schema
spark.read.format("delta").load("/tmp/test-delta-table").printSchema()

# 检查版本兼容性
spark-shell --version
# 选择对应的 Delta Lake 版本

<!-- 排除冲突依赖 -->
<dependency>
    <groupId>io.delta</groupId>
    <artifactId>delta-core_2.12</artifactId>
    <version>2.4.0</version>
    <exclusions>
        <exclusion>
            <groupId>conflicting-group</groupId>
            <artifactId>conflicting-artifact</artifactId>
        </exclusion>
    </exclusions>
</dependency>

# 检查存储权限
spark.sql("CREATE TABLE test_permissions (id INT) USING DELTA LOCATION '/path/to/test'")

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 创建示例数据
data = [
    (1, "Alice", 25, "Engineering"),
    (2, "Bob", 30, "Sales"),
    (3, "Charlie", 35, "Marketing")
]
columns = ["id", "name", "age", "department"]
df = spark.createDataFrame(data, columns)

# 写入 Delta Table
df.write.format("delta").save("/path/to/delta-table")

# 带分区的 Delta Table
df.write.format("delta") \
    .partitionBy("department") \
    .save("/path/to/partitioned-table")

import org.apache.spark.sql.SparkSession
import io.delta.tables._

val data = Seq(
  (1, "Alice", 25, "Engineering"),
  (2, "Bob", 30, "Sales"),
  (3, "Charlie", 35, "Marketing")
)

val df = spark.createDataFrame(data).toDF("id", "name", "age", "department")

// 写入 Delta Table
df.write.format("delta").save("/path/to/delta-table")

-- 创建 Delta Table
CREATE TABLE employees (
    id INT,
    name STRING,
    age INT,
    department STRING,
    salary DOUBLE,
    hire_date DATE
) USING DELTA
LOCATION '/path/to/delta-table'

-- 创建分区表
CREATE TABLE employees_partitioned (
    id INT,
    name STRING,
    age INT,
    department STRING,
    salary DOUBLE,
    hire_date DATE
) USING DELTA
PARTITIONED BY (department)
LOCATION '/path/to/partitioned-table'

-- 创建带约束的表
CREATE TABLE employees_with_constraints (
    id INT NOT NULL,
    name STRING NOT NULL,
    age INT CHECK (age >= 18 AND age <= 65),
    department STRING,
    salary DOUBLE CHECK (salary > 0)
) USING DELTA

# 从 Parquet 文件创建
parquet_df = spark.read.parquet("/path/to/parquet-files")
parquet_df.write.format("delta").save("/path/to/delta-table")

# 从 CSV 文件创建
csv_df = spark.read.option("header", "true").csv("/path/to/csv-files")
csv_df.write.format("delta").save("/path/to/delta-table")

# 从数据库表创建
jdbc_df = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydb") \
    .option("dbtable", "source_table") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

jdbc_df.write.format("delta").save("/path/to/delta-table")

# 读取整个 Delta Table
df = spark.read.format("delta").load("/path/to/delta-table")
df.show()

# 使用 SQL 读取
spark.sql("SELECT * FROM delta.`/path/to/delta-table`").show()

# 注册为临时视图
spark.read.format("delta").load("/path/to/delta-table").createOrReplaceTempView("employees")
spark.sql("SELECT * FROM employees WHERE age > 30").show()

# 使用 DataFrame API
df = spark.read.format("delta").load("/path/to/delta-table")
filtered_df = df.filter(df.age > 30).select("name", "department", "age")
filtered_df.show()

# 使用 SQL
result = spark.sql("""
    SELECT name, department, age 
    FROM delta.`/path/to/delta-table` 
    WHERE age > 30 AND department = 'Engineering'
""")
result.show()

# 部门统计
dept_stats = spark.sql("""
    SELECT 
        department,
        COUNT(*) as employee_count,
        AVG(age) as avg_age,
        MAX(salary) as max_salary,
        MIN(salary) as min_salary
    FROM delta.`/path/to/delta-table`
    GROUP BY department
    ORDER BY employee_count DESC
""")
dept_stats.show()

# 准备新数据
new_data = [
    (4, "David", 28, "Engineering"),
    (5, "Eve", 32, "Sales"),
    (6, "Frank", 29, "Marketing")
]
new_df = spark.createDataFrame(new_data, columns)

# 追加到 Delta Table
new_df.write.format("delta").mode("append").save("/path/to/delta-table")

# 使用 SQL 插入
spark.sql("""
    INSERT INTO delta.`/path/to/delta-table`
    VALUES (7, 'Grace', 27, 'Engineering'),
           (8, 'Henry', 31, 'Sales')
""")

# 大批量数据插入
large_df = spark.range(10000).select(
    col("id").cast("int"),
    concat(lit("User"), col("id")).alias("name"),
    (col("id") % 50 + 20).alias("age"),
    when(col("id") % 3 == 0, "Engineering")
    .when(col("id") % 3 == 1, "Sales")
    .otherwise("Marketing").alias("department")
)

# 优化写入性能
large_df.coalesce(10) \
    .write.format("delta") \
    .mode("append") \
    .save("/path/to/delta-table")

from delta.tables import DeltaTable

# 加载 Delta Table
deltaTable = DeltaTable.forPath(spark, "/path/to/delta-table")

# 简单更新
deltaTable.update(
    condition = "age < 30",
    set = {"age": "age + 1"}
)

# 复杂更新
deltaTable.update(
    condition = "department = 'Engineering' AND age > 35",
    set = {
        "salary": "salary * 1.1",
        "last_updated": "current_timestamp()"
    }
)

-- 简单更新
UPDATE delta.`/path/to/delta-table`
SET age = age + 1
WHERE age < 30

-- 条件更新
UPDATE delta.`/path/to/delta-table`
SET salary = salary * 1.1,
    last_updated = current_timestamp()
WHERE department = 'Engineering' AND performance_rating = 'Excellent'

-- 基于子查询更新
UPDATE delta.`/path/to/delta-table` t1
SET salary = (
    SELECT AVG(salary) * 1.2 
    FROM delta.`/path/to/salary-reference` t2 
    WHERE t2.department = t1.department
)
WHERE t1.salary < (
    SELECT AVG(salary) 
    FROM delta.`/path/to/salary-reference` t3 
    WHERE t3.department = t1.department
)

# 使用 Delta Table API
deltaTable.delete("age > 60")

# 删除特定部门
deltaTable.delete("department = 'Marketing' AND hire_date < '2020-01-01'")

-- 简单删除
DELETE FROM delta.`/path/to/delta-table`
WHERE age > 60

-- 复杂删除条件
DELETE FROM delta.`/path/to/delta-table`
WHERE department = 'Sales' 
  AND hire_date < '2020-01-01'
  AND performance_rating = 'Poor'

-- 基于子查询删除
DELETE FROM delta.`/path/to/delta-table` t1
WHERE EXISTS (
    SELECT 1 FROM delta.`/path/to/terminated-employees` t2
    WHERE t2.employee_id = t1.id
)

# 准备更新数据
updates = [
    (1, "Alice Smith", 26, "Engineering"),  # 更新现有记录
    (9, "Ivy", 29, "Engineering"),          # 插入新记录
    (10, "Jack", 33, "Sales")               # 插入新记录
]
updates_df = spark.createDataFrame(updates, columns)

# 执行 Merge 操作
deltaTable.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set = {
    "name": "source.name",
    "age": "source.age",
    "department": "source.department"
}).whenNotMatchedInsert(values = {
    "id": "source.id",
    "name": "source.name",
    "age": "source.age",
    "department": "source.department"
}).execute()

# 复杂的 Merge 逻辑
deltaTable.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    condition = "source.age IS NOT NULL",
    set = {
        "name": "source.name",
        "age": "source.age",
        "department": "source.department",
        "last_updated": "current_timestamp()"
    }
).whenMatchedDelete(
    condition = "source.status = 'TERMINATED'"
).whenNotMatchedInsert(
    condition = "source.age >= 18",
    values = {
        "id": "source.id",
        "name": "source.name",
        "age": "source.age",
        "department": "source.department",
        "created_at": "current_timestamp()"
    }
).execute()

MERGE INTO delta.`/path/to/delta-table` AS target
USING (
    SELECT id, name, age, department, status
    FROM delta.`/path/to/updates-table`
) AS source
ON target.id = source.id

WHEN MATCHED AND source.status = 'ACTIVE' THEN
    UPDATE SET 
        name = source.name,
        age = source.age,
        department = source.department,
        last_updated = current_timestamp()

WHEN MATCHED AND source.status = 'TERMINATED' THEN
    DELETE

WHEN NOT MATCHED AND source.age >= 18 THEN
    INSERT (id, name, age, department, created_at)
    VALUES (source.id, source.name, source.age, source.department, current_timestamp())

# 查看表结构
spark.sql("DESCRIBE delta.`/path/to/delta-table`").show()

# 查看详细信息
spark.sql("DESCRIBE DETAIL delta.`/path/to/delta-table`").show()

# 查看表历史
deltaTable.history().show()

# 查看表属性
spark.sql("SHOW TBLPROPERTIES delta.`/path/to/delta-table`").show()

# 生成表统计信息
spark.sql("ANALYZE TABLE delta.`/path/to/delta-table` COMPUTE STATISTICS")

# 刷新表缓存
spark.sql("REFRESH TABLE delta.`/path/to/delta-table`")

# 修复表（如果需要）
spark.sql("MSCK REPAIR TABLE delta.`/path/to/delta-table`")

# 事务中的多个操作
try:
    # 开始事务（隐式）
    deltaTable.update(condition="department = 'Sales'", set={"salary": "salary * 1.1"})
    deltaTable.delete("performance_rating = 'Poor'")
    
    # 插入新数据
    new_employees.write.format("delta").mode("append").save("/path/to/delta-table")
    
    # 事务自动提交
    print("事务执行成功")
    
except Exception as e:
    print(f"事务执行失败: {e}")
    # Delta Lake 自动回滚

# 配置重试机制
spark.conf.set("spark.databricks.delta.retryWriteConflict.enabled", "true")
spark.conf.set("spark.databricks.delta.retryWriteConflict.limit", "3")

# 并发安全的写入
def safe_write(df, path, max_retries=3):
    for attempt in range(max_retries):
        try:
            df.write.format("delta").mode("append").save(path)
            print(f"写入成功，尝试次数: {attempt + 1}")
            break
        except Exception as e:
            if "ConcurrentModificationException" in str(e) and attempt < max_retries - 1:
                print(f"写入冲突，重试中... (尝试 {attempt + 1})")
                time.sleep(2 ** attempt)  # 指数退避
            else:
                raise e

# 查询特定版本
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/path/to/delta-table")
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("/path/to/delta-table")

# 使用 SQL 查询版本
spark.sql("SELECT * FROM delta.`/path/to/delta-table` VERSION AS OF 3").show()

# 比较不同版本
current_df = spark.read.format("delta").load("/path/to/delta-table")
previous_df = spark.read.format("delta").option("versionAsOf", 5).load("/path/to/delta-table")

# 找出差异
diff_df = current_df.subtract(previous_df)
diff_df.show()

# 查询特定时间点
df_timestamp = spark.read.format("delta") \
    .option("timestampAsOf", "2023-01-01 12:00:00") \
    .load("/path/to/delta-table")

# 使用 SQL 时间戳查询
spark.sql("""
    SELECT * FROM delta.`/path/to/delta-table` 
    TIMESTAMP AS OF '2023-01-01 12:00:00'
""").show()

# 查询一小时前的数据
from datetime import datetime, timedelta
one_hour_ago = (datetime.now() - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
df_hour_ago = spark.read.format("delta") \
    .option("timestampAsOf", one_hour_ago) \
    .load("/path/to/delta-table")

# 查看完整历史
history_df = deltaTable.history()
history_df.select("version", "timestamp", "operation", "operationParameters").show(truncate=False)

# 分析版本变化
version_stats = history_df.groupBy("operation").count().orderBy("count", ascending=False)
version_stats.show()

# 查看特定操作的历史
merge_history = history_df.filter("operation = 'MERGE'")
merge_history.select("version", "timestamp", "operationParameters.predicate").show(truncate=False)

# 回滚到版本 5
deltaTable.restoreToVersion(5)

# 使用 SQL 回滚
spark.sql("RESTORE TABLE delta.`/path/to/delta-table` TO VERSION AS OF 5")

# 验证回滚结果
current_version = deltaTable.history().first()["version"]
print(f"当前版本: {current_version}")

# 回滚到特定时间
deltaTable.restoreToTimestamp("2023-01-01 12:00:00")

# 使用 SQL 时间点回滚
spark.sql("""
    RESTORE TABLE delta.`/path/to/delta-table` 
    TO TIMESTAMP AS OF '2023-01-01 12:00:00'
""")

def safe_restore(table_path, target_version, backup_path=None):
    """安全的表回滚操作"""
    delta_table = DeltaTable.forPath(spark, table_path)
    
    # 记录当前状态
    current_version = delta_table.history().first()["version"]
    print(f"当前版本: {current_version}")
    
    # 可选：创建备份
    if backup_path:
        current_df = spark.read.format("delta").load(table_path)
        current_df.write.format("delta").save(backup_path)
        print(f"备份已创建: {backup_path}")
    
    # 执行回滚
    try:
        delta_table.restoreToVersion(target_version)
        print(f"成功回滚到版本: {target_version}")
        
        # 验证回滚
        new_version = delta_table.history().first()["version"]
        print(f"回滚后版本: {new_version}")
        
    except Exception as e:
        print(f"回滚失败: {e}")
        raise

# 手动触发压缩
deltaTable.optimize().executeCompaction()

# 使用 SQL 压缩
spark.sql("OPTIMIZE delta.`/path/to/delta-table`")

# 压缩特定分区
spark.sql("OPTIMIZE delta.`/path/to/delta-table` WHERE department = 'Engineering'")

# 对常用查询列进行 Z-Order 优化
deltaTable.optimize().executeZOrderBy("department", "age")

# 使用 SQL 进行 Z-Order
spark.sql("OPTIMIZE delta.`/path/to/delta-table` ZORDER BY (department, age)")

# 多列 Z-Order 优化
deltaTable.optimize().executeZOrderBy("department", "age", "hire_date", "salary")

# 启用自动优化
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# 配置优化参数
spark.conf.set("spark.databricks.delta.autoCompact.minNumFiles", "50")
spark.conf.set("spark.databricks.delta.autoCompact.maxFileSize", "1073741824")  # 1GB

# 清理 7 天前的历史文件
deltaTable.vacuum(168)  # 168 小时 = 7 天

# 使用 SQL 清理
spark.sql("VACUUM delta.`/path/to/delta-table` RETAIN 168 HOURS")

# 清理特定分区
spark.sql("VACUUM delta.`/path/to/delta-table` RETAIN 168 HOURS WHERE department = 'Engineering'")

def safe_vacuum(table_path, retention_hours=168, dry_run=True):
    """安全的历史数据清理"""
    delta_table = DeltaTable.forPath(spark, table_path)
    
    if dry_run:
        # 预览将要删除的文件
        print("预览模式 - 将要删除的文件:")
        spark.conf.set("spark.databricks.delta.vacuum.logging.enabled", "true")
        delta_table.vacuum(retention_hours)
    else:
        # 实际执行清理
        print(f"清理 {retention_hours} 小时前的历史文件...")
        delta_table.vacuum(retention_hours)
        print("清理完成")

# 先预览，再执行
safe_vacuum("/path/to/delta-table", retention_hours=168, dry_run=True)
safe_vacuum("/path/to/delta-table", retention_hours=168, dry_run=False)

# 配置默认保留期
spark.conf.set("spark.databricks.delta.deletedFileRetentionDuration", "interval 3 days")
spark.conf.set("spark.databricks.delta.logRetentionDuration", "interval 7 days")

# 禁用安全检查（谨慎使用）
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

# 基本流式读取
stream = spark.readStream.format("delta").load("/path/to/delta-table")

# 从特定版本开始读取
stream_from_version = spark.readStream \
    .format("delta") \
    .option("startingVersion", "5") \
    .load("/path/to/delta-table")

# 从特定时间开始读取
stream_from_time = spark.readStream \
    .format("delta") \
    .option("startingTimestamp", "2023-01-01 00:00:00") \
    .load("/path/to/delta-table")

# 基本流式写入
query = stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/path/to/checkpoint") \
    .start("/path/to/output-table")

# 流式 Upsert
def upsert_to_delta(batch_df, batch_id):
    """流式批次的 Upsert 操作"""
    delta_table = DeltaTable.forPath(spark, "/path/to/target-table")
    
    delta_table.alias("target").merge(
        batch_df.alias("source"),
        "target.id = source.id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

# 应用 Upsert 到流
upsert_query = stream.writeStream \
    .foreachBatch(upsert_to_delta) \
    .option("checkpointLocation", "/path/to/checkpoint") \
    .start()

# 实时聚合统计
windowed_counts = stream \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "5 minutes"),
        col("department")
    ).agg(
        count("*").alias("employee_count"),
        avg("age").alias("avg_age")
    )

# 写入聚合结果
aggregation_query = windowed_counts.writeStream \
    .format("delta") \
    .outputMode("update") \
    .option("checkpointLocation", "/path/to/agg-checkpoint") \
    .start("/path/to/aggregation-table")

# 为新表启用 CDF
spark.sql("""
    CREATE TABLE cdf_enabled_table (
        id INT,
        name STRING,
        age INT
    ) USING DELTA
    TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# 为现有表启用 CDF
spark.sql("""
    ALTER TABLE delta.`/path/to/delta-table`
    SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

# 读取从特定版本开始的变更
changes_df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 0) \
    .load("/path/to/delta-table")

changes_df.show()

# 读取特定时间范围的变更
time_range_changes = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingTimestamp", "2023-01-01 00:00:00") \
    .option("endingTimestamp", "2023-01-02 00:00:00") \
    .load("/path/to/delta-table")

# 流式读取变更
change_stream = spark.readStream.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 0) \
    .load("/path/to/delta-table")

# 处理不同类型的变更
def process_changes(batch_df, batch_id):
    """处理变更数据"""
    inserts = batch_df.filter("_change_type = 'insert'")
    updates = batch_df.filter("_change_type = 'update_postimage'")
    deletes = batch_df.filter("_change_type = 'delete'")
    
    print(f"批次 {batch_id}: 插入 {inserts.count()}, 更新 {updates.count()}, 删除 {deletes.count()}")
    
    # 处理各种变更类型
    if inserts.count() > 0:
        # 处理插入
        pass
    if updates.count() > 0:
        # 处理更新
        pass
    if deletes.count() > 0:
        # 处理删除
        pass

# 启动变更处理流
change_query = change_stream.writeStream \
    .foreachBatch(process_changes) \
    .option("checkpointLocation", "/path/to/cdf-checkpoint") \
    .start()

-- 创建浅克隆（只复制元数据）
CREATE TABLE shallow_clone
SHALLOW CLONE delta.`/path/to/source-table`

-- 克隆到特定版本
CREATE TABLE version_clone
SHALLOW CLONE delta.`/path/to/source-table` VERSION AS OF 5

-- 创建深克隆（复制所有数据）
CREATE TABLE deep_clone
DEEP CLONE delta.`/path/to/source-table`

-- 克隆到特定时间点
CREATE TABLE timestamp_clone
DEEP CLONE delta.`/path/to/source-table` TIMESTAMP AS OF '2023-01-01 12:00:00'

# 启用自动优化写入
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# 配置文件大小参数
spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728")  # 128MB
spark.conf.set("spark.databricks.delta.autoCompact.minNumFiles", "50")
spark.conf.set("spark.databricks.delta.autoCompact.maxFileSize", "1073741824")  # 1GB

# 定期执行压缩
deltaTable.optimize().executeCompaction()

# 监控文件分布
file_stats = spark.sql("""
    DESCRIBE DETAIL delta.`/path/to/delta-table`
""").select("numFiles", "sizeInBytes")
file_stats.show()

# 检查小文件数量
small_files_count = spark.sql("""
    SELECT COUNT(*) as small_files_count
    FROM (
        DESCRIBE DETAIL delta.`/path/to/delta-table`
    ) WHERE sizeInBytes < 134217728
""")

# 控制写入分区数
optimal_partitions = max(1, df.count() // 1000000)  # 每分区约100万行
df.coalesce(optimal_partitions).write.format("delta").save("/path/to/table")

# 批量写入优化
df.write.format("delta") \
    .option("maxRecordsPerFile", 1000000) \
    .mode("append") \
    .save("/path/to/table")

# 查看统计信息
spark.sql("""
    DESCRIBE DETAIL delta.`/path/to/delta-table`
""").select("minValues", "maxValues", "nullCount").show(truncate=False)

# 利用数据跳过的查询
efficient_query = spark.sql("""
    SELECT * FROM delta.`/path/to/delta-table`
    WHERE age BETWEEN 25 AND 35
    AND department = 'Engineering'
""")

# 分析分区效果
partition_stats = spark.sql("""
    SELECT 
        department,
        COUNT(*) as record_count,
        COUNT(DISTINCT date) as date_range
    FROM delta.`/path/to/partitioned-table`
    GROUP BY department
""")
partition_stats.show()

# 重新分区优化
df.write.format("delta") \
    .partitionBy("year", "month") \
    .save("/path/to/optimized-table")

# 选择合适的 Z-Order 列
# 1. 经常一起查询的列
# 2. 高基数列
# 3. 范围查询列

deltaTable.optimize().executeZOrderBy("department", "age", "salary")

# 监控 Z-Order 效果
before_stats = spark.sql("SELECT COUNT(*) FROM delta.`/path/to/table` WHERE department='Engineering' AND age > 30")
# 执行 Z-Order
deltaTable.optimize().executeZOrderBy("department", "age")
after_stats = spark.sql("SELECT COUNT(*) FROM delta.`/path/to/table` WHERE department='Engineering' AND age > 30")

# 缓存整个表
deltaTable.cache()

# 缓存查询结果
frequent_query = spark.sql("""
    SELECT department, AVG(salary) as avg_salary
    FROM delta.`/path/to/delta-table`
    GROUP BY department
""")
frequent_query.cache()
frequent_query.count()  # 触发缓存

def smart_cache_strategy(table_path, query_patterns):
    """智能缓存策略"""
    delta_table = DeltaTable.forPath(spark, table_path)
    
    # 分析表大小
    table_stats = spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").collect()[0]
    table_size_mb = table_stats["sizeInBytes"] / (1024 * 1024)
    
    if table_size_mb < 1000:  # 小于1GB的表
        # 缓存整个表
        delta_table.cache()
        print("缓存整个表")
    else:
        # 缓存热点分区
        for pattern in query_patterns:
            hot_partition = spark.sql(f"""
                SELECT * FROM delta.`{table_path}` 
                WHERE {pattern['condition']}
            """)
            hot_partition.cache()
            hot_partition.count()  # 触发缓存
            print(f"缓存热点分区: {pattern['condition']}")

# 应用智能缓存
query_patterns = [
    {"condition": "department = 'Engineering'"},
    {"condition": "hire_date >= '2023-01-01'"}
]
smart_cache_strategy("/path/to/delta-table", query_patterns)

# 配置并发参数
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")  # 128MB

# 优化并发写入
spark.conf.set("spark.databricks.delta.merge.repartitionBeforeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.retryWriteConflict.enabled", "true")

# 分区级别的并发控制
def partition_aware_write(df, table_path, partition_col):
    """分区感知的并发写入"""
    partitions = df.select(partition_col).distinct().collect()
    
    for partition in partitions:
        partition_value = partition[partition_col]
        partition_df = df.filter(f"{partition_col} = '{partition_value}'")
        
        # 每个分区独立写入，减少冲突
        partition_df.write.format("delta") \
            .mode("append") \
            .option("replaceWhere", f"{partition_col} = '{partition_value}'") \
            .save(table_path)

# 配置压缩算法
spark.conf.set("spark.sql.parquet.compression.codec", "snappy")  # 平衡压缩率和速度
# spark.conf.set("spark.sql.parquet.compression.codec", "gzip")   # 高压缩率
# spark.conf.set("spark.sql.parquet.compression.codec", "lz4")    # 高速度

# 列级压缩优化
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")
spark.conf.set("spark.sql.parquet.columnarReaderBatchSize", "4096")

# 优化 Parquet 写入
spark.conf.set("spark.sql.parquet.writeLegacyFormat", "false")
spark.conf.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
spark.conf.set("spark.sql.parquet.int96RebaseModeInWrite", "CORRECTED")

# 字典编码优化
spark.conf.set("spark.sql.parquet.enableDictionary", "true")
spark.conf.set("spark.sql.parquet.dictionaryPageSize", "1048576")  # 1MB

def auto_optimize_table(table_path, optimization_config):
    """自动化表优化"""
    delta_table = DeltaTable.forPath(spark, table_path)
    
    # 获取表统计信息
    table_detail = spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").collect()[0]
    num_files = table_detail["numFiles"]
    size_in_bytes = table_detail["sizeInBytes"]
    
    # 判断是否需要压缩
    if num_files > optimization_config["max_files"]:
        print(f"文件数量过多 ({num_files})，执行压缩...")
        delta_table.optimize().executeCompaction()
    
    # 判断是否需要 Z-Order
    if size_in_bytes > optimization_config["zorder_threshold"]:
        print("表较大，执行 Z-Order 优化...")
        delta_table.optimize().executeZOrderBy(*optimization_config["zorder_columns"])
    
    # 判断是否需要清理
    history_count = delta_table.history().count()
    if history_count > optimization_config["max_versions"]:
        print("版本过多，执行清理...")
        delta_table.vacuum(optimization_config["retention_hours"])

# 配置优化参数
config = {
    "max_files": 1000,
    "zorder_threshold": 10 * 1024 * 1024 * 1024,  # 10GB
    "zorder_columns": ["department", "age"],
    "max_versions": 100,
    "retention_hours": 168  # 7天
}

# 执行自动优化
auto_optimize_table("/path/to/delta-table", config)

def benchmark_query_performance(table_path, queries, iterations=3):
    """查询性能基准测试"""
    results = []
    
    for query_name, query_sql in queries.items():
        times = []
        
        for i in range(iterations):
            start_time = time.time()
            
            # 执行查询
            result_df = spark.sql(query_sql.format(table_path=table_path))
            result_count = result_df.count()
            
            end_time = time.time()
            execution_time = end_time - start_time
            times.append(execution_time)
            
            print(f"{query_name} - 迭代 {i+1}: {execution_time:.2f}秒, 结果数: {result_count}")
        
        avg_time = sum(times) / len(times)
        results.append({
            "query": query_name,
            "avg_time": avg_time,
            "min_time": min(times),
            "max_time": max(times)
        })
    
    return results

# 定义基准查询
benchmark_queries = {
    "full_scan": "SELECT COUNT(*) FROM delta.`{table_path}`",
    "filtered_scan": "SELECT * FROM delta.`{table_path}` WHERE department = 'Engineering'",
    "aggregation": "SELECT department, AVG(salary) FROM delta.`{table_path}` GROUP BY department",
    "join_query": """
        SELECT t1.*, t2.bonus 
        FROM delta.`{table_path}` t1 
        JOIN delta.`/path/to/bonus-table` t2 ON t1.id = t2.employee_id
    """
}

# 执行基准测试
performance_results = benchmark_query_performance("/path/to/delta-table", benchmark_queries)

# 分析结果
for result in performance_results:
    print(f"查询: {result['query']}, 平均时间: {result['avg_time']:.2f}秒")

# 推荐的数据类型映射
data_type_mapping = {
    "整数ID": "BIGINT",          # 而不是 INT，为未来扩展预留空间
    "金额": "DECIMAL(18,2)",      # 精确的货币计算
    "时间戳": "TIMESTAMP",        # 而不是 STRING
    "布尔值": "BOOLEAN",          # 而不是 STRING('Y'/'N')
    "枚举值": "STRING",           # 使用约束限制取值
    "JSON数据": "STRING"          # 存储为字符串，查询时解析
}

# 示例表结构
spark.sql("""
    CREATE TABLE optimized_employees (
        employee_id BIGINT NOT NULL,
        name STRING NOT NULL,
        email STRING NOT NULL,
        age INT CHECK (age >= 18 AND age <= 65),
        salary DECIMAL(18,2) CHECK (salary > 0),
        department STRING NOT NULL,
        hire_date DATE NOT NULL,
        is_active BOOLEAN DEFAULT true,
        metadata STRING,  -- JSON格式的扩展信息
        created_at TIMESTAMP DEFAULT current_timestamp(),
        updated_at TIMESTAMP DEFAULT current_timestamp()
    ) USING DELTA
    PARTITIONED BY (department, date_format(hire_date, 'yyyy-MM'))
    TBLPROPERTIES (
        'delta.enableChangeDataFeed' = 'true',
        'delta.autoOptimize.optimizeWrite' = 'true',
        'delta.autoOptimize.autoCompact' = 'true'
    )
""")

# 分区策略示例
def design_partition_strategy(df, business_requirements):
    """设计分区策略"""
    
    # 分析数据分布
    partition_analysis = df.groupBy("department", "year", "month").count()
    partition_analysis.show()
    
    # 评估分区效果
    total_partitions = partition_analysis.count()
    avg_records_per_partition = df.count() / total_partitions
    
    print(f"总分区数: {total_partitions}")
    print(f"平均每分区记录数: {avg_records_per_partition}")
    
    # 推荐分区策略
    if total_partitions > 10000:
        print("建议: 减少分区粒度，避免小文件问题")
    elif avg_records_per_partition < 100000:
        print("建议: 增加分区粒度，提高查询效率")
    else:
        print("分区策略合理")

# 应用分区策略分析
design_partition_strategy(df, {"query_pattern": "time_based", "data_retention": "2_years"})

# 推荐的命名规范
naming_conventions = {
    "原始数据表": "raw_{source_system}_{table_name}",
    "清洗后数据": "clean_{domain}_{table_name}",
    "聚合表": "agg_{domain}_{granularity}_{table_name}",
    "维度表": "dim_{dimension_name}",
    "事实表": "fact_{business_process}",
    "临时表": "temp_{process_name}_{timestamp}"
}

# 示例
table_names = [
    "raw_crm_customers",           # 原始CRM客户数据
    "clean_sales_orders",          # 清洗后的销售订单
    "agg_sales_daily_revenue",     # 日粒度销售收入聚合
    "dim_product_catalog",         # 产品目录维度表
    "fact_customer_transactions",  # 客户交易事实表
    "temp_data_migration_20231201" # 临时数据迁移表
]

class DeltaDataQualityFramework:
    """Delta Lake 数据质量管理框架"""
    
    def __init__(self, spark_session, table_path):
        self.spark = spark_session
        self.table_path = table_path
        self.delta_table = DeltaTable.forPath(spark_session, table_path)
    
    def add_constraints(self, constraints):
        """添加数据约束"""
        for constraint_name, constraint_condition in constraints.items():
            try:
                self.spark.sql(f"""
                    ALTER TABLE delta.`{self.table_path}`
                    ADD CONSTRAINT {constraint_name} CHECK ({constraint_condition})
                """)
                print(f"约束 {constraint_name} 添加成功")
            except Exception as e:
                print(f"约束 {constraint_name} 添加失败: {e}")
    
    def validate_data_quality(self, df):
        """数据质量验证"""
        quality_report = {}
        
        # 完整性检查
        total_records = df.count()
        quality_report["total_records"] = total_records
        
        # 空值检查
        null_counts = {}
        for column in df.columns:
            null_count = df.filter(col(column).isNull()).count()
            null_percentage = (null_count / total_records) * 100
            null_counts[column] = {
                "null_count": null_count,
                "null_percentage": round(null_percentage, 2)
            }
        quality_report["null_analysis"] = null_counts
        
        # 重复数据检查
        duplicate_count = total_records - df.dropDuplicates().count()
        quality_report["duplicate_count"] = duplicate_count
        quality_report["duplicate_percentage"] = round((duplicate_count / total_records) * 100, 2)
        
        # 数据分布检查
        for column in df.select_dtypes(include=['int', 'double', 'float']).columns:
            stats = df.select(column).describe().collect()
            quality_report[f"{column}_stats"] = {row["summary"]: row[column] for row in stats}
        
        return quality_report
    
    def auto_fix_quality_issues(self, df, fix_config):
        """自动修复数据质量问题"""
        fixed_df = df
        
        # 处理空值
        if "fill_nulls" in fix_config:
            for column, fill_value in fix_config["fill_nulls"].items():
                fixed_df = fixed_df.fillna({column: fill_value})
        
        # 去重
        if fix_config.get("remove_duplicates", False):
            fixed_df = fixed_df.dropDuplicates()
        
        # 数据类型转换
        if "type_conversions" in fix_config:
            for column, target_type in fix_config["type_conversions"].items():
                fixed_df = fixed_df.withColumn(column, col(column).cast(target_type))
        
        return fixed_df

# 使用示例
quality_framework = DeltaDataQualityFramework(spark, "/path/to/delta-table")

# 添加约束
constraints = {
    "age_range": "age >= 18 AND age <= 65",
    "salary_positive": "salary > 0",
    "email_format": "email LIKE '%@%.%'"
}
quality_framework.add_constraints(constraints)

# 数据质量检查
df = spark.read.format("delta").load("/path/to/delta-table")
quality_report = quality_framework.validate_data_quality(df)
print(json.dumps(quality_report, indent=2))

class DataLineageTracker:
    """数据血缘追踪器"""
    
    def __init__(self, spark_session):
        self.spark = spark_session
        self.lineage_table = "/path/to/lineage-metadata"
    
    def record_transformation(self, source_tables, target_table, transformation_type, metadata):
        """记录数据转换血缘"""
        lineage_record = spark.createDataFrame([{
            "transformation_id": str(uuid.uuid4()),
            "source_tables": source_tables,
            "target_table": target_table,
            "transformation_type": transformation_type,
            "metadata": json.dumps(metadata),
            "created_at": datetime.now(),
            "created_by": getpass.getuser()
        }])
        
        lineage_record.write.format("delta").mode("append").save(self.lineage_table)
    
    def track_delta_operation(self, table_path, operation_type, operation_metadata):
        """追踪 Delta 操作"""
        # 记录操作到用户元数据
        user_metadata = {
            "operation_type": operation_type,
            "timestamp": datetime.now().isoformat(),
            "user": getpass.getuser(),
            "metadata": operation_metadata
        }
        
        return json.dumps(user_metadata)

# 使用示例
lineage_tracker = DataLineageTracker(spark)

# 记录 ETL 转换
lineage_tracker.record_transformation(
    source_tables=["raw_sales", "dim_products"],
    target_table="fact_sales_summary",
    transformation_type="aggregation",
    metadata={
        "aggregation_level": "daily",
        "metrics": ["total_sales", "order_count"],
        "filters": "order_date >= '2023-01-01'"
    }
)

# 在写入时记录血缘
metadata = lineage_tracker.track_delta_operation(
    "/path/to/target-table",
    "merge",
    {"source": "daily_batch", "records_processed": 10000}
)

df.write.format("delta") \
    .option("userMetadata", metadata) \
    .mode("append") \
    .save("/path/to/target-table")

class DeltaTableMaintenanceManager:
    """Delta 表维护管理器"""
    
    def __init__(self, spark_session, config):
        self.spark = spark_session
        self.config = config
        self.maintenance_log = "/path/to/maintenance-log"
    
    def daily_maintenance(self, table_paths):
        """日常维护任务"""
        maintenance_results = []
        
        for table_path in table_paths:
            try:
                result = self._maintain_single_table(table_path)
                maintenance_results.append(result)
                
                # 记录维护日志
                self._log_maintenance(table_path, result)
                
            except Exception as e:
                error_result = {
                    "table_path": table_path,
                    "status": "failed",
                    "error": str(e),
                    "timestamp": datetime.now()
                }
                maintenance_results.append(error_result)
                self._log_maintenance(table_path, error_result)
        
        return maintenance_results
    
    def _maintain_single_table(self, table_path):
        """单表维护"""
        delta_table = DeltaTable.forPath(self.spark, table_path)
        result = {"table_path": table_path, "actions": []}
        
        # 获取表统计信息
        table_detail = self.spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").collect()[0]
        num_files = table_detail["numFiles"]
        size_in_bytes = table_detail["sizeInBytes"]
        
        # 检查是否需要压缩
        if num_files > self.config["compaction_threshold"]:
            delta_table.optimize().executeCompaction()
            result["actions"].append("compaction")
        
        # 检查是否需要 Z-Order
        if size_in_bytes > self.config["zorder_threshold"]:
            zorder_columns = self.config.get("zorder_columns", {}).get(table_path, [])
            if zorder_columns:
                delta_table.optimize().executeZOrderBy(*zorder_columns)
                result["actions"].append(f"zorder: {zorder_columns}")
        
        # 检查是否需要清理历史版本
        history_count = delta_table.history().count()
        if history_count > self.config["max_versions"]:
            delta_table.vacuum(self.config["retention_hours"])
            result["actions"].append("vacuum")
        
        result["status"] = "success"
        result["timestamp"] = datetime.now()
        return result
    
    def _log_maintenance(self, table_path, result):
        """记录维护日志"""
        log_df = self.spark.createDataFrame([result])
        log_df.write.format("delta").mode("append").save(self.maintenance_log)
    
    def generate_maintenance_report(self, start_date, end_date):
        """生成维护报告"""
        report_df = self.spark.sql(f"""
            SELECT 
                table_path,
                status,
                actions,
                COUNT(*) as maintenance_count,
                MAX(timestamp) as last_maintenance
            FROM delta.`{self.maintenance_log}`
            WHERE timestamp BETWEEN '{start_date}' AND '{end_date}'
            GROUP BY table_path, status, actions
            ORDER BY table_path, last_maintenance DESC
        """)
        
        return report_df

# 配置维护管理器
maintenance_config = {
    "compaction_threshold": 1000,
    "zorder_threshold": 10 * 1024 * 1024 * 1024,  # 10GB
    "max_versions": 100,
    "retention_hours": 168,  # 7天
    "zorder_columns": {
        "/path/to/sales-table": ["department", "date"],
        "/path/to/customer-table": ["region", "signup_date"]
    }
}

maintenance_manager = DeltaTableMaintenanceManager(spark, maintenance_config)

# 执行日常维护
tables_to_maintain = [
    "/path/to/sales-table",
    "/path/to/customer-table",
    "/path/to/product-table"
]

maintenance_results = maintenance_manager.daily_maintenance(tables_to_maintain)

# 生成维护报告
report = maintenance_manager.generate_maintenance_report("2023-01-01", "2023-01-31")
report.show()

class DeltaTableMonitor:
    """Delta 表监控系统"""
    
    def __init__(self, spark_session, alert_config):
        self.spark = spark_session
        self.alert_config = alert_config
        self.metrics_table = "/path/to/table-metrics"
    
    def collect_table_metrics(self, table_paths):
        """收集表指标"""
        all_metrics = []
        
        for table_path in table_paths:
            try:
                metrics = self._get_table_metrics(table_path)
                all_metrics.append(metrics)
            except Exception as e:
                print(f"收集 {table_path} 指标失败: {e}")
        
        # 保存指标
        if all_metrics:
            metrics_df = self.spark.createDataFrame(all_metrics)
            metrics_df.write.format("delta").mode("append").save(self.metrics_table)
        
        return all_metrics
    
    def _get_table_metrics(self, table_path):
        """获取单表指标"""
        delta_table = DeltaTable.forPath(self.spark, table_path)
        
        # 基础指标
        table_detail = self.spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").collect()[0]
        
        # 历史指标
        history = delta_table.history().limit(10).collect()
        recent_operations = [row["operation"] for row in history]
        
        # 数据新鲜度
        latest_timestamp = history[0]["timestamp"] if history else None
        data_freshness_hours = None
        if latest_timestamp:
            data_freshness_hours = (datetime.now() - latest_timestamp).total_seconds() / 3600
        
        return {
            "table_path": table_path,
            "num_files": table_detail["numFiles"],
            "size_gb": table_detail["sizeInBytes"] / (1024**3),
            "num_versions": len(history),
            "data_freshness_hours": data_freshness_hours,
            "recent_operations": recent_operations,
            "collected_at": datetime.now()
        }
    
    def check_alerts(self, metrics):
        """检查告警条件"""
        alerts = []
        
        for metric in metrics:
            table_path = metric["table_path"]
            
            # 文件数量告警
            if metric["num_files"] > self.alert_config["max_files"]:
                alerts.append({
                    "table": table_path,
                    "type": "too_many_files",
                    "value": metric["num_files"],
                    "threshold": self.alert_config["max_files"],
                    "severity": "warning"
                })
            
            # 数据新鲜度告警
            if metric["data_freshness_hours"] and metric["data_freshness_hours"] > self.alert_config["max_freshness_hours"]:
                alerts.append({
                    "table": table_path,
                    "type": "stale_data",
                    "value": metric["data_freshness_hours"],
                    "threshold": self.alert_config["max_freshness_hours"],
                    "severity": "critical"
                })
            
            # 表大小告警
            if metric["size_gb"] > self.alert_config["max_size_gb"]:
                alerts.append({
                    "table": table_path,
                    "type": "large_table",
                    "value": metric["size_gb"],
                    "threshold": self.alert_config["max_size_gb"],
                    "severity": "info"
                })
        
        return alerts
    
    def send_alerts(self, alerts):
        """发送告警"""
        if not alerts:
            return
        
        # 按严重程度分组
        critical_alerts = [a for a in alerts if a["severity"] == "critical"]
        warning_alerts = [a for a in alerts if a["severity"] == "warning"]
        info_alerts = [a for a in alerts if a["severity"] == "info"]
        
        # 发送告警（这里只是打印，实际可以集成邮件、Slack等）
        if critical_alerts:
            print("🚨 严重告警:")
            for alert in critical_alerts:
                print(f"  - {alert['table']}: {alert['type']} = {alert['value']} (阈值: {alert['threshold']})")
        
        if warning_alerts:
            print("⚠️ 警告告警:")
            for alert in warning_alerts:
                print(f"  - {alert['table']}: {alert['type']} = {alert['value']} (阈值: {alert['threshold']})")
        
        if info_alerts:
            print("ℹ️ 信息告警:")
            for alert in info_alerts:
                print(f"  - {alert['table']}: {alert['type']} = {alert['value']} (阈值: {alert['threshold']})")

# 配置监控系统
alert_config = {
    "max_files": 5000,
    "max_freshness_hours": 24,
    "max_size_gb": 1000
}

monitor = DeltaTableMonitor(spark, alert_config)

# 收集指标并检查告警
tables_to_monitor = ["/path/to/table1", "/path/to/table2"]
metrics = monitor.collect_table_metrics(tables_to_monitor)
alerts = monitor.check_alerts(metrics)
monitor.send_alerts(alerts)