如何在Databricks中通过Blob存储文件生成PySpark表Schema

实现步骤： #

1. 读取Blob存储中的文本文件
   在Databricks中，Blob存储的路径格式为wasbs://<容器名>@<存储账户名>.blob.core.windows.net/<文件路径>，使用Spark文本读取API加载文件内容。

2. 解析每行的表名与列名
   文件每行格式为表名=['列1','列2',...]，通过正则表达式提取表名和列列表，可按需处理重复列。

3. 生成StructType Schema
   遍历解析后的表与列，为每个列创建StructField（类型为StringType，允许为空），组合成StructType后存入字典，方便后续调用。

完整代码示例： #

# 导入必要的类型模块
from pyspark.sql.types import StructType, StructField, StringType
import re

# 1. 读取Blob存储中的文本文件
file_path = "wasbs://your-container@your-storage-account.blob.core.windows.net/path/to/your/file.txt"
lines_df = spark.read.text(file_path)
lines = [row.value for row in lines_df.collect()]

# 2. 解析表名和列名，生成schema字典
schema_dict = {}
for line in lines:
    # 用正则匹配表名和列列表部分
    match = re.match(r"(\w+)\s*=\s*\[(.+)\]", line.strip())
    if match:
        table_name = match.group(1)
        cols_str = match.group(2)
        # 拆分列名并去除引号、空白字符，可选去重
        columns = [col.strip().strip("'\"") for col in cols_str.split(",")]
        columns = list(dict.fromkeys(columns))  # 去重，保留首次出现的列
        
        # 3. 生成对应的StructType
        struct_fields = [StructField(col, StringType(), nullable=True) for col in columns]
        schema_dict[table_name] = StructType(struct_fields)

# 调用示例：获取对应表的schema
table1_schema = schema_dict.get("table1")
table2_schema = schema_dict.get("table2")

# 验证schema结构
print(table1_schema)
print(table2_schema)

说明： #

• 路径替换：将file_path中的容器名、存储账户名和文件路径替换为实际值。
• 重复列处理：代码中加入了列名去重逻辑，若无需去重，删除columns = list(dict.fromkeys(columns))即可。
• schema存储：所有表的schema存入schema_dict，通过表名可快速获取对应的StructType。

内容的提问来源于stack exchange，提问作者Swati B