如何使用sparklyr直接读取Blob存储中的CSV文件无需下载？

1. 配置Spark连接并集成Azure Blob认证 #

要让sparklyr直接读取Azure Blob中的文件，需在Spark初始化时添加Blob存储的认证配置，让Spark直接通过配置访问存储，无需下载文件：

library(sparklyr)
library(AzureStor)

# 你的Blob存储凭证
account_endpoint <- "https://mycorporation.blob.core.windows.net"
account_key      <- "mykey"
container_name   <- "mycorporation"
folder_path      <- "my_folder"

# 初始化Spark连接，添加Azure Blob配置
sc <- spark_connect(
  master = "local", # 根据实际Spark集群配置修改，如YARN、K8s等
  config = list(
    spark.hadoop.fs.azure.account.key.mycorporation.blob.core.windows.net = account_key,
    spark.hadoop.fs.azure = "org.apache.hadoop.fs.azure.NativeAzureFileSystem"
  )
)

2. 获取目标文件夹的CSV文件列表（按顺序） #

用AzureStor列出目标文件夹下的所有CSV文件，并按需求排序（比如文件名升序，保证读取顺序）：

# 建立Azure存储端点和容器连接
bl_endp_key <- storage_endpoint(account_endpoint, key = account_key)
cont        <- storage_container(bl_endp_key, container_name)

# 列出my_folder下的所有CSV文件，过滤非CSV文件
blob_files <- list_blobs(cont, prefix = folder_path, recursive = FALSE)
csv_files <- blob_files[sapply(blob_files, function(x) grepl("\\.csv$", x$name))]

# 按文件名排序，确保读取顺序
csv_files_sorted <- csv_files[order(sapply(csv_files, function(x) x$name))]

# 转换为Spark可识别的Blob路径格式：wasbs://容器名@存储账户名.blob.core.windows.net/文件路径
spark_file_paths <- sapply(csv_files_sorted, function(x) {
  paste0("wasbs://", container_name, "@mycorporation.blob.core.windows.net/", x$name)
})

3. 按顺序读取并处理CSV文件 #

根据需求选择合并为单个DataFrame或逐个处理文件：

方式1：合并为单个Spark DataFrame #

# 初始化空的Spark DataFrame
combined_df <- NULL

for (path in spark_file_paths) {
  # 读取当前CSV文件
  current_df <- spark_read_csv(sc, path = path, header = TRUE, infer_schema = TRUE)
  
  # 合并到总DataFrame
  if (is.null(combined_df)) {
    combined_df <- current_df
  } else {
    combined_df <- combined_df %>% sdf_union_all(current_df)
  }
}

方式2：逐个处理文件（按需操作） #

for (path in spark_file_paths) {
  current_df <- spark_read_csv(sc, path = path, header = TRUE, infer_schema = TRUE)
  
  # 在这里添加自定义处理逻辑，比如数据清洗、统计分析等
  # 示例：打印当前文件的行数
  cat("处理文件：", path, "，行数：", sdf_nrow(current_df), "\n")
}

关键说明 #

• 使用wasbs://协议让Spark直接访问Azure Blob存储，无需将文件下载到本地或集群节点
• 排序文件列表时可根据实际需求调整排序规则（比如修改order()的参数）
• 若已知CSV列类型，可手动指定schema参数替代infer_schema = TRUE，提升读取性能
• 若使用集群模式（如YARN），需确保集群节点能访问Azure Blob存储，且认证配置正确

内容的提问来源于stack exchange，提问作者Rodrigo H. Ozon