如何在Azure Databricks中上传Spark Dataframe至Azure Table Storage并建表？

一、前提准备 #

• 确认你的Service Principal拥有Azure Table Storage的Contributor或Storage Table Data Contributor权限
• 补充Service Principal的client_secret（你提到的server_id实际应为tenant ID）
• 确保Databricks集群已安装必要依赖：若用Spark原生连接器，需添加Maven包com.microsoft.azure:azure-storage-spark:2.0.0；若用Python SDK，可在Notebook中执行%pip install azure-data-tables安装依赖

二、将Spark DataFrame上传至Azure Table Storage #

推荐使用Spark原生连接器适配分布式场景，两种方式如下：

方式1：Spark原生连接器（优先选择） #

# 1. 配置基础参数
storage_account_name = "<你的存储账户名>"
target_table = "<目标表名>"
client_id = "<你的client_id>"
tenant_id = "<你的server_id（即tenant ID）>"
client_secret = "<你的Service Principal密钥>"

# 2. 设置Spark连接配置
spark.conf.set(f"fs.azure.account.auth.type.{storage_account_name}.table.core.windows.net", "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{storage_account_name}.table.core.windows.net", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{storage_account_name}.table.core.windows.net", client_id)
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{storage_account_name}.table.core.windows.net", client_secret)
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{storage_account_name}.table.core.windows.net", f"https://login.microsoftonline.com/{tenant_id}/oauth2/token")

# 3. 写入DataFrame（需确保DataFrame包含PartitionKey和RowKey字段）
df.write \
  .format("azure-table") \
  .option("tableName", target_table) \
  .option("storageAccount", storage_account_name) \
  .mode("append")  # 可选模式：append/overwrite/ignore/errorifexists
  .save()

方式2：Python SDK结合Spark（仅适合小数据量） #

%pip install azure-data-tables

from azure.data.tables import TableServiceClient
from azure.identity import ClientSecretCredential

# 1. 初始化Table服务客户端
storage_account_name = "<你的存储账户名>"
target_table = "<目标表名>"
credential = ClientSecretCredential(tenant_id="<你的server_id>", client_id="<你的client_id>", client_secret="<你的client_secret>")
table_service = TableServiceClient(endpoint=f"https://{storage_account_name}.table.core.windows.net", credential=credential)

# 2. Spark DataFrame转Pandas（仅小数据量适用）
pandas_df = df.toPandas()

# 3. 批量写入实体
with table_service.get_table_client(target_table) as table_client:
    entities = [
        {
            "PartitionKey": row["PartitionKey"],
            "RowKey": row["RowKey"],
            **row.to_dict()
        }
        for _, row in pandas_df.iterrows()
    ]
    table_client.upsert_entity(entities)

三、基于Spark DataFrame创建Azure Table Storage表 #

Azure Table Storage是无SQL存储，无需预先定义表结构：首次写入数据时，若目标表不存在会自动创建，只需确保DataFrame包含PartitionKey和RowKey这两个必填主键字段。

若需提前创建空表，可通过Python SDK实现：

from azure.identity import ClientSecretCredential
from azure.data.tables import TableServiceClient

storage_account_name = "<你的存储账户名>"
target_table = "<目标表名>"
credential = ClientSecretCredential(tenant_id="<你的server_id>", client_id="<你的client_id>", client_secret="<你的client_secret>")
table_service = TableServiceClient(endpoint=f"https://{storage_account_name}.table.core.windows.net", credential=credential)

# 创建空表（若已存在则跳过）
table_service.create_table_if_not_exists(target_table)

常见问题排查 #

• 连接失败：检查Service Principal权限是否正确，确认tenant_id、client_id、client_secret无拼写错误
• 写入报错：确保DataFrame包含PartitionKey和RowKey字段，这是Azure Table Storage的强制主键
• 性能问题：大数据量场景务必使用Spark原生连接器，避免转为Pandas DataFrame

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