如何用Python向Azure Blob Storage的CSV文件追加不重复行？

由于Azure Blob存储的特性：Block Blob仅支持整文件替换，Append Blob只能追加内容无法修改或去重已有数据，所以常规最优方案就是先下载远程CSV，和本地数据合并去重后再重新上传。

具体实现步骤 #

1. 下载远程Blob中的CSV内容
2. 解析本地和远程的CSV数据，以ID为唯一键去重（保留双方不重复的行，避免ID重复）
3. 将合并去重后的新CSV上传覆盖原Blob

代码示例 #

from azure.storage.blob import ContainerClient
import csv
import io

# 初始化容器客户端（沿用你原有的配置）
container_client = ContainerClient(account_url=url, container_name=name, credential=creds)
blob_name = "your-file-name.csv"

# 1. 获取远程CSV数据
remote_rows = {}
try:
    blob_client = container_client.get_blob_client(blob_name)
    # 下载Blob内容到内存
    remote_content = blob_client.download_blob().readall().decode('utf-8')
    # 解析远程CSV，用ID作为键存储行数据
    reader = csv.DictReader(io.StringIO(remote_content))
    for row in reader:
        remote_rows[row['ID']] = row
except Exception as e:
    # 如果远程文件不存在，直接用本地数据
    print(f"远程文件不存在或读取失败: {e}")

# 2. 解析本地CSV数据（假设你的data是本地CSV的字符串/字节流）
local_content = data.decode('utf-8') if isinstance(data, bytes) else data
local_reader = csv.DictReader(io.StringIO(local_content))
# 合并数据：保留远程已有的ID，新增本地独有的ID
for row in local_reader:
    if row['ID'] not in remote_rows:
        remote_rows[row['ID']] = row

# 3. 生成去重后的新CSV
output = io.StringIO()
writer = csv.DictWriter(output, fieldnames=local_reader.fieldnames)
writer.writeheader()
writer.writerows(remote_rows.values())

# 4. 上传覆盖原Blob
container_client.upload_blob(name=blob_name, data=output.getvalue(), overwrite=True)

额外说明 #

• 如果需要更新已有ID对应的value（比如本地的value比远程新），可以把合并逻辑改成：不管ID是否存在，都用本地行覆盖远程行，直接删除if row['ID'] not in remote_rows:判断即可。
• 如果CSV文件极大，内存不够用，可以把数据写到临时本地文件，而非内存中的StringIO，处理完成后再上传。

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