如何在Azure Databricks中批量替换Excel内JSON的用户哈希值为随机掩码

问题描述 #

File Store中有一个Excel文件，包含超过10000列JSON数据，每个JSON的Users字段包含用户哈希值与邮箱地址。需要将这些用户邮箱批量替换为指定列表中的随机掩码邮箱（例如Alvin.Baltado@ph.vroonshipmanagement.com替换为sam@contoso.com），此前手动替换耗时极长，希望通过Azure Databricks实现一次性批量处理并将结果写回指定位置。

此前手动处理的示例代码：

import random
main=['nzn1@contoso.com', 'oman2@contoso.com', 'oman3@contoso.com', 'oman4@contoso.com', 'oman5@contoso.com', 'oman6@contoso.com', 'oman7@contoso.com', 'oman8@contoso.com', 'oman9@contoso.com', 'omaz1@contoso.com', 'omaz2@contoso.com', 'omaz3@contoso.com', 'omaz4@contoso.com', 'omaz5@contoso.com', 'omaom6@contoso.com', 'omax7@contoso.com', 'omaz8@contoso.com', 'omaz9@contoso.com', 'omay1@contoso.com', 'omay2@contoso.com', 'omaom3@contoso.com', 'omax4@contoso.com', 'omax5@contoso.com', 'omax6@contoso.com', 'omaom7@contoso.com', 'omaw8@contoso.com', 'omaw9@contoso.com', 'omae1@contoso.com', 'omae2@contoso.com', 'omae3@contoso.com', 'omae4@contoso.com', 'omae5@contoso.com']  

l=["0e07209b-807b-4938-8bfd-f87cee98e924,invoices@it.vroonoffshore.com,c747a82c-656e-40eb-9194-88c4a0f8061e"]
n=len(l)
print(n)
print(random.sample(main,n))

解决方案 #

步骤1：环境准备与数据读取 #

在Azure Databricks中使用PySpark处理，需先确保集群安装了spark-excel依赖（通过集群库管理添加com.crealytics:spark-excel_2.12:0.13.5，版本需匹配Spark版本）。

步骤2：定义替换逻辑与批量处理 #

编写UDF解析JSON字段，随机替换Users中的邮箱地址，对所有JSON列批量应用该逻辑。

步骤3：结果写入 #

将处理后的DataFrame写回指定存储位置。

完整实现代码：

from pyspark.sql.functions import udf, col
from pyspark.sql.types import StringType
import json
import random

# 读取Excel文件
input_path = "/FileStore/path/to/your/source/file.xlsx"
df = spark.read.format("com.crealytics.spark.excel") \
    .option("header", "true") \
    .option("inferSchema", "false") \
    .load(input_path)

# 定义掩码邮箱列表
mask_emails = [
    'nzn1@contoso.com', 'oman2@contoso.com', 'oman3@contoso.com', 
    'oman4@contoso.com', 'oman5@contoso.com', 'oman6@contoso.com', 
    'oman7@contoso.com', 'oman8@contoso.com', 'oman9@contoso.com', 
    'omaz1@contoso.com', 'omaz2@contoso.com', 'omaz3@contoso.com', 
    'omaz4@contoso.com', 'omaz5@contoso.com', 'omaom6@contoso.com', 
    'omax7@contoso.com', 'omaz8@contoso.com', 'omaz9@contoso.com', 
    'omay1@contoso.com', 'omay2@contoso.com', 'omaom3@contoso.com', 
    'omax4@contoso.com', 'omax5@contoso.com', 'omax6@contoso.com', 
    'omaom7@contoso.com', 'omaw8@contoso.com', 'omaw9@contoso.com', 
    'omae1@contoso.com', 'omae2@contoso.com', 'omae3@contoso.com', 
    'omae4@contoso.com', 'omae5@contoso.com'
]

# 定义邮箱替换UDF（需根据实际JSON结构调整）
def mask_user_json(json_str):
    try:
        data = json.loads(json_str)
        if "Users" in data:
            random_email = random.choice(mask_emails)
            # 适配Users为字典的情况（示例结构：{"hash": "...", "email": "..."}）
            if isinstance(data["Users"], dict) and "email" in data["Users"]:
                data["Users"]["email"] = random_email
            # 适配Users为列表的情况（示例结构：[{"hash": "...", "email": "..."}, ...]）
            elif isinstance(data["Users"], list):
                for user in data["Users"]:
                    if "email" in user:
                        user["email"] = random_email
        return json.dumps(data)
    except Exception:
        return json_str

mask_udf = udf(mask_user_json, StringType())

# 筛选所有字符串类型列（假设为JSON列），批量应用替换
json_columns = [col_name for col_name in df.columns if df.schema[col_name].dataType == StringType()]
for col_name in json_columns:
    df = df.withColumn(col_name, mask_udf(col(col_name)))

# 写入处理后的Excel文件
output_path = "/FileStore/path/to/your/output/masked_file.xlsx"
df.write.format("com.crealytics.spark.excel") \
    .option("header", "true") \
    .mode("overwrite") \
    .save(output_path)

print("批量处理完成，结果已写入:", output_path)

注意事项 #

• JSON结构适配：需根据实际Users字段的JSON格式调整UDF中的解析逻辑
• 性能优化：针对超大量列，可通过Spark分区配置提升并行处理效率
• 依赖版本：spark-excel版本需与集群Spark版本匹配，避免兼容性问题

内容的提问来源于stack exchange，提问作者data Greek