Azure Databricks中Python实现流式Delta表整表加密解密方案问询

针对你希望无需逐列操作、实现整个Delta表加密解密的需求，以下是三种可行方案：

方案1：批量处理所有目标列（自定义加密逻辑） #

基于你现有的列级加密代码，通过遍历表中所有需要加密的列（比如所有字符串类型列），批量应用加密/解密UDF，避免手动逐列指定。

代码示例 #

from pyspark.sql.functions import udf, lit, col
from pyspark.sql.types import StringType
from cryptography.fernet import Fernet

# 定义加密解密函数
def encrypt_val(value, key):
    if value is None:
        return None
    f = Fernet(key)
    return f.encrypt(value.encode()).decode()

def decrypt_val(value, key):
    if value is None:
        return None
    f = Fernet(key)
    return f.decrypt(value.encode()).decode()

# 注册UDF
encrypt_udf = udf(encrypt_val, StringType())
decrypt_udf = udf(decrypt_val, StringType())

# 从Databricks Secrets获取密钥（替换为你的密钥范围和名称）
encryption_key = dbutils.secrets.get(scope="encrypt", key="fernetkey")

# 批量加密所有字符串类型列
df = spark.table("EncryptTest")
# 筛选出所有字符串类型的列
string_columns = [col_name for col_name, dtype in df.dtypes if dtype == "string"]

# 循环应用加密
encrypted_df = df
for col_name in string_columns:
    encrypted_df = encrypted_df.withColumn(col_name, encrypt_udf(col(col_name), lit(encryption_key)))

# 保存为加密Delta表
encrypted_df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").saveAsTable("Encryption_Test_Table")

# 批量解密验证
decrypted_df = encrypted_df
for col_name in string_columns:
    decrypted_df = decrypted_df.withColumn(col_name, decrypt_udf(col(col_name), lit(encryption_key)))

display(decrypted_df)

方案2：Delta Lake内置透明数据加密（TDE） #

如果不需要自定义加密算法，推荐使用Delta Lake的透明数据加密，这是存储层的自动加密，无需手动处理列，流式写入也会自动生效，加密覆盖整个表的数据和元数据。

创建加密Delta表（SQL方式） #

CREATE TABLE Encrypted_Stream_Table (
  id INT,
  ssn STRING,
  address STRING
)
USING DELTA
LOCATION "/dbfs/path/to/encrypted_table"
TBLPROPERTIES (delta.encrypted = true);

Python流式写入加密表 #

# 读取流式数据源（以Kafka为例）
streaming_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-server:9092") \
    .option("subscribe", "source-topic") \
    .load()

# 解析数据并写入加密Delta表
parsed_stream = streaming_df.selectExpr("CAST(value AS STRING)") \
    .select(from_json(col("value"), your_schema).alias("data")) \
    .select("data.*")

# 流式写入加密表
write_query = parsed_stream.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/dbfs/checkpoints/stream_checkpoint") \
    .table("Encrypted_Stream_Table")

方案3：流式Delta表批量加密处理 #

针对流式场景，在流处理链路中批量应用自定义加密逻辑，确保每一批次的数据都自动完成加密。

代码示例 #

# 定义流式读取源
streaming_source = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .load("/dbfs/streaming_source")

# 获取需要加密的列（字符串类型）
string_cols = [col_name for col_name, dtype in streaming_source.dtypes if dtype == "string"]

# 批量加密
encrypted_stream = streaming_source
for col_name in string_cols:
    encrypted_stream = encrypted_stream.withColumn(col_name, encrypt_udf(col(col_name), lit(encryption_key)))

# 写入流式Delta表
stream_query = encrypted_stream.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/dbfs/checkpoints/encrypted_stream") \
    .start("/dbfs/streaming_encrypted_delta")

关键注意事项 #

• 密钥管理：务必使用Databricks Secrets结合Azure Key Vault存储加密密钥，禁止硬编码密钥。
• UDF序列化：自定义加密函数需确保可序列化，避免分布式计算时出现异常。
• 流式检查点：流式任务必须指定持久化的checkpoint位置（如DBFS路径），确保任务中断后可恢复。
• TDE依赖：使用Delta内置TDE需确保Databricks工作区和底层存储账户已配置相应加密权限。

内容的提问来源于stack exchange，提问作者Rajshri Dewangan