如何从Blob Storage读取Personal_details_info系列文件到Databricks Notebook

1. 配置Blob Storage访问连接 #

首先需为Databricks配置Blob Storage的访问权限，以下是使用SAS令牌的示例（也可使用服务principal方式）：

# 替换为你的存储账户信息
storage_account_name = "your_storage_account"
container_name = "your_container"
sas_token = "your_sas_token"

# 设置Spark连接配置
spark.conf.set(f"fs.azure.sas.{container_name}.{storage_account_name}.blob.core.windows.net", sas_token)

2. 读取Personal_detail_Info系列文件 #

利用通配符*匹配所有同系列文件，无需逐个指定文件名：

# 读取文件，根据实际格式调整format参数（如csv/parquet/json）
personal_df = spark.read \
    .format("csv")
    .option("header", "true")  # 文件含表头则启用
    .option("inferSchema", "true")  # 自动推断Schema，生产环境建议手动指定
    .load(f"wasbs://{container_name}@{storage_account_name}.blob.core.windows.net/Personal_detail_Info_*")

# 预览前5行数据
personal_df.show(5)

3. 读取Customer_details_Info系列文件 #

采用同样的通配符方式读取另一系列文件：

customer_df = spark.read \
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load(f"wasbs://{container_name}@{storage_account_name}.blob.core.windows.net/Customer_details_Info_*")

# 预览数据
customer_df.show(5)

关键注意事项 #

• 生产环境建议：避免使用inferSchema，手动定义Schema可提升性能并避免类型推断错误，示例如下：

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

# 自定义Personal_detail_Info的Schema
personal_schema = StructType([
    StructField("user_id", StringType(), nullable=False),
    StructField("full_name", StringType(), nullable=True),
    StructField("birth_date", DateType(), nullable=True),
    StructField("phone_number", StringType(), nullable=True)
])

# 读取时指定Schema
personal_df = spark.read \
    .format("csv")
    .option("header", "true")
    .schema(personal_schema)
    .load(f"wasbs://{container_name}@{storage_account_name}.blob.core.windows.net/Personal_detail_Info_*")

• 格式适配：若文件为Parquet、JSON等格式，只需将format参数替换为对应值即可。

后续数据处理示例 #

若需合并两个数据集（假设存在共同关联键user_id）：

# 内连接两个表
combined_user_df = personal_df.join(customer_df, on="user_id", how="inner")

# 保存处理结果到Blob Storage
combined_user_df.write \
    .format("parquet")
    .mode("overwrite")
    .save(f"wasbs://{container_name}@{storage_account_name}.blob.core.windows.net/combined_user_details")

内容的提问来源于stack exchange，提问作者bigdata techie