如何将Blob Storage中的CSV文件推送至Azure Cosmos DB for Apache Cassandra？

方法一：Azure Data Factory（ADF）批量导入 #

适合无代码/低代码的批量数据迁移场景：

• 配置Blob数据源：在ADF中创建Blob Storage链接服务，填入存储账户连接信息；创建CSV数据集，指定存储CSV的容器路径及文件格式（如分隔符、编码）。
• 配置Cassandra目标：创建Azure Cosmos DB for Apache Cassandra链接服务，填入集群端点、访问密钥、目标密钥空间名称；创建Cassandra数据集，关联到目标表。
• 构建映射数据流：新建映射数据流，源选择Blob CSV数据集，目标选择Cassandra数据集，在字段映射面板中匹配CSV列与Cassandra表字段（需确保数据类型一致，例如CSV文本对应Cassandra的varchar，数字对应int/bigint）。
• 执行导入：创建管道并添加映射数据流活动，手动触发或设置定时触发，完成数据导入。

方法二：Python脚本自定义导入 #

适合需要自定义数据清洗或特殊逻辑的场景：

1. 安装依赖包：

pip install azure-storage-blob cassandra-driver pandas

2. 编写核心逻辑示例：

from azure.storage.blob import BlobServiceClient
import pandas as pd
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider

# Blob Storage配置
BLOB_CONN_STR = "你的Blob存储连接字符串"
CONTAINER_NAME = "存储CSV的容器名"
BLOB_NAME = "目标CSV文件名.csv"

# Cosmos DB Cassandra配置
CASSANDRA_ENDPOINT = "你的Cosmos DB Cassandra集群端点"
CASSANDRA_KEY = "你的Cosmos DB访问密钥"
KEYSPACE = "目标密钥空间名"
TABLE = "目标表名"

# 从Blob读取CSV到内存
blob_service = BlobServiceClient.from_connection_string(BLOB_CONN_STR)
blob_client = blob_service.get_blob_client(container=CONTAINER_NAME, blob=BLOB_NAME)
csv_content = blob_client.download_blob().readall().decode('utf-8')
df = pd.read_csv(pd.io.common.StringIO(csv_content))

# 连接Cassandra集群
auth = PlainTextAuthProvider(username='cassandra', password=CASSANDRA_KEY)
cluster = Cluster([CASSANDRA_ENDPOINT], port=10350, auth_provider=auth, ssl=True)
session = cluster.connect(KEYSPACE)

# 批量插入数据
insert_stmt = session.prepare(f"INSERT INTO {TABLE} (列名1, 列名2, 列名3) VALUES (?, ?, ?)")
for _, row in df.iterrows():
    session.execute(insert_stmt, (row['CSV列名1'], row['CSV列名2'], row['CSV列名3']))

# 清理连接
session.shutdown()
cluster.shutdown()

关键注意事项 #

• 提前在Cosmos DB Cassandra中创建好对应的密钥空间和表，确保字段类型与CSV数据匹配，避免导入失败。
• 若CSV数据量极大，ADF的分布式处理更稳定；Python脚本可通过多线程/批量提交优化插入速度。
• 确保操作主体（ADF服务主体、Python脚本）拥有Blob Storage的读取权限及Cosmos DB Cassandra的写入权限。

内容的提问来源于stack exchange，提问作者Simran Pal Singh