对象存储
对象存储
对象存储
文档指南
请输入
产品动态
产品公告
产品计费
快速入门
用户指南
存储桶管理
基础设置
容错与容灾
ObjectSet
文件管理
监控与告警
权限配置
IAM 策略管理
典型配置场景
IAM 子用户授权
使用工具及 SDK 的授权说明
接入点
单区域接入点（SRAP）
内容感知与处理（SenseFlow）
TOS Agent
数据处理
分层命名空间（HNS）
TOS Vectors
API 参考
向量桶
SDK 参考
Python-Vectors
向量桶
向量索引
Java-Vectors
向量桶
向量索引
Node.js-Vectors
向量桶
向量索引
工具指南
tosutil
常用命令
辅助命令
TOS Browser
FSX
TOS PyTorch Connector
tosfsspec
API 参考
签名机制
存储桶
桶基础操作
自定义域名
数据处理
媒体处理
视频转码
音频转码
ObjectSet
生命周期
对象
基础操作
分片上传
SDK 参考
Java
存储桶接口
管理桶配置
ObjectSet
对象接口
下载对象
数据处理
视频处理
日志配置
分层命名空间
Python
数据处理
存储桶接口
对象接口
上传对象
C++
存储桶接口
对象接口
Go
对象接口
存储桶接口
管理桶配置
ObjectSet
数据处理
日志配置
Node.js
存储桶接口
对象接口
上传对象
数据处理
Browser.js
上传对象
.NET
对象接口
PHP
对象接口
异常处理
Android
对象接口
iOS
对象接口
Harmony
存储桶接口
对象接口
上传对象
Rust
存储桶接口
桶基础操作
桶清单
自定义域名
对象接口
基础操作
错误码
01-SERVICE_ERROR
02-AUTH
03-ACCESS_CONTROL
04-QOS
05-REQUEST
06-BUCKET
07-LIFECYCLE
08-ACL
09-CORS
10-POLICY
11-MIRROR
12-REPLICATION
13-WEBSITE
14-NOTIFICATION
15-ENCRYPT
16-CUSTOMDOMAIN
17-OBJECT
18-TAG
19-MULTIPART
20-STORAGE_CLASS
21-CALLBACK
22-DATA_PROCESS
23-BATCH
24-STORAGE_LENS
25-INVENTORY
26-DIRECTORY_BUCKET
32-WORM
常见问题
最佳实践
AIGC-豆包大模型最佳实践
校验数据一致性
- 文档首页
对象存储工具指南tosfsspectosfsspec 生态集成
复制全文
下载 pdftosfsspec 生态集成
文档反馈
问问助手tosfsspec 完整实现了 fsspec 规范，除可直接在 Python 中使用外，还能与 Pandas、Dask、Ray、PyArrow、PyTorch 等主流数据分析与 AI 框架无缝集成。
说明
在 fsspec 中使用
注意
您可直接通过 fsspec.open() 访问 TOS 资源：在 URL 中使用 tos-new:// 协议，并通过 storage_options 传入所需的构造参数即可。
代码示例
import fsspec
import os
import tos
fsspec.register_implementation('tos-new', 'tosfsspec.TosFileSystem')
url = 'tos-new://your-bucket/path/to/data.csv'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
with fsspec.open(url, 'rb', **storage_options) as f:
head = f.read(1024)
print(head)
fsspec.register_implementation用法下，fsspec 会自动选择 TosFileSystem 作为底层文件系统实现。
在 Python 中使用
在 Python 中可直接通过 tosfsspec 的常用 API 读写 TOS，类似本地文件系统的使用方式。
代码示例
import os
import tosfsspec
import tos
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
fs = tosfsspec.TosFileSystem(
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
region='cn-beijing'
)
bucket = 'your-bucket'
local_file_path = 'localfile.txt'
remote_file_path = f'tos://{bucket}/remote_file.txt'
# create a local file to upload
with open(local_file_path, 'w') as f:
f.write('Hello TOSFS.')
# upload to tos
fs.put_file(local_file_path, remote_file_path)
print(f'Uploaded {local_file_path} to {remote_file_path}')
# download from tos
downloaded_file_path = 'downloaded_file.txt'
fs.get_file(remote_file_path, downloaded_file_path)
print(f'Downloaded {remote_file_path} to {downloaded_file_path}')
# read content from downloaded local file
with open(downloaded_file_path, 'r') as f:
content = f.read()
print(f'Content of {downloaded_file_path}: {content}')
# delete tos file
fs.rm(remote_file_path)
# write to tos
with fs.open(remote_file_path, 'w') as f:
f.write('Hello TOSFS.')
# read from tos
with fs.open(remote_file_path, 'r') as f:
tos_content = f.read()
print(f'Content of {remote_file_path}: {tos_content}')
# clean local files
os.remove(local_file_path)
os.remove(downloaded_file_path)
在多进程环境下使用
tosfsspec 已适配多进程场景，可在多进程环境下使用；但仍建议以 spawn 模式启动子进程。
代码示例
import multiprocessing
import tosfsspec
import os
import tos
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
def action(fs):
result = fs.ls(f'tos://{bucket}/')
print(result)
if __name__ == '__main__':
# 推荐使用 spawn 模式
multiprocessing.set_start_method('spawn', force=True)
fs = tosfsspec.TosFileSystem(
key=key,
secret=secret,
region='cn-beijing'
)
processes = [multiprocessing.Process(target=action, args=(fs,)) for i in range(4)]
for p in processes:
p.start()
for p in processes:
p.join()
与 Pandas 集成
tosfsspec 遵循 fsspec 规范，因此 Pandas 可借助 tosfsspec 直接从 TOS 读取或写入 CSV、Parquet 等多种格式数据，无需先下载到本地。
- 从 TOS 读取 CSV 文件
代码示例
import fsspec
import pandas as pd
import tosfsspec
import os
import tos
fsspec.register_implementation('tos-new', 'tosfsspec.TosFileSystem')
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
# 准备一个 CSV 文件在 TOS 上
fs = tosfsspec.TosFileSystem(
key=key,
secret=secret,
region='cn-beijing'
)
bucket = 'your-bucket'
csv_path = f'tos-new://{bucket}/datasets/iris.csv'
csv_content = 'sepal_length,sepal_width,petal_length,petal_width,species\n5.1,3.5,1.4,0.2,setosa\n'
with fs.open(csv_path, 'wb') as f:
f.write(csv_content.encode())
# 配置 storage_options
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
# 使用 pandas 直接读取
df = pd.read_csv(csv_path, storage_options=storage_options)
print('成功从 TOS 读取数据:')
print(df)
- 将 DataFrame 写入 TOS 的 Parquet 文件
代码示例
import fsspec
import pandas as pd
import os
import tos
fsspec.register_implementation('tos-new', 'tosfsspec.TosFileSystem')
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
# 创建一个 DataFrame
data = {'col1': [1, 2], 'col2': [3, 4]}
df_to_write = pd.DataFrame(data=data)
# 定义要写入的 Parquet 文件路径
parquet_path = f'tos-new://{bucket}/output/data.parquet'
# 配置 storage_options
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
# 直接写入 TOS，需要安装 pyarrow 或 fastparquet
df_to_write.to_parquet(parquet_path, storage_options=storage_options, index=False)
print(f'DataFrame 已成功写入到: {parquet_path}')
与 Dask 集成
Dask 是一套灵活的并行计算库，天然支持 fsspec 兼容的文件系统。结合 tosfsspec 后，可通过 dask.dataframe 一次性并行读取 TOS 上的多个 CSV 文件，并构建分布式 DataFrame。
- 假设您的 TOS Bucket 中存放了如下结构的多个 CSV 文件：
代码示例
your-bucket/
└── stock_data/
├── part.0.csv
├── part.1.csv
└── part.2.csv
- 您可以使用 dask 一次性将它们全部加载进来：
代码示例
import dask.dataframe as dd
import fsspec
import os
import tos
fsspec.register_implementation('tos-new', 'tosfsspec.TosFileSystem')
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
# 配置 storage_options
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
# 文件路径支持通配符
dask_path = f'tos-new://{bucket}/stock_data/part.*.csv'
# 使用 Dask 读取数据，Dask 会自动处理通配符并并行加载
ddf = dd.read_csv(dask_path, storage_options=storage_options)
# Dask 是惰性计算的，此时数据还未真正加载
print('Dask DataFrame 分区数:', ddf.npartitions)
# 执行计算，例如计算平均值，这会触发实际的 I/O 操作
mean_value = ddf.some_column.mean().compute()
print('计算得到的平均值:', mean_value)
与 Ray 分布式计算集成
Ray 可结合 tosfsspec 在分布式集群中直接读取并处理 TOS 上的数据。
代码示例
import ray
import tosfsspec
import os
import tos
# 初始化 Ray (通常在集群环境中已初始化)
ray.init()
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
fs = tosfsspec.TosFileSystem(
key=key,
secret=secret,
region='cn-beijing'
)
# csv file path
input_csv_path = f'tos://{bucket}/input.csv'
output_csv_path = f'tos://{bucket}/output.csv'
with fs.open(input_csv_path, 'w') as f:
f.write('id,name,age\n1,John Doe,30\n2,Jane Smith,25\n3,Bob Johnson,40\n')
ds = ray.data.read_csv(input_csv_path, filesystem=fs)
# processing data
# ds = ds.map_batches(your_processing_function)
ds.repartition(1).write_csv(output_csv_path, filesystem=fs)
与 PyArrow 集成
pyarrow可通过 filesystem 参数使用 tosfsspec，实现高效读写 TOS 中的数据。
代码示例
import pyarrow as pa
import pyarrow.dataset as ds
import pyarrow.parquet as pq
import tosfsspec
import os
import tos
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
fs = tosfsspec.TosFileSystem(
key=key,
secret=secret,
region='cn-beijing'
)
# 生成 Parquet 数据集
table = pa.table({'year': [2020, 2022, 2021, 2022, 2019, 2021],
'n_legs': [2, 2, 4, 4, 5, 100],
'animal': ['Flamingo', 'Parrot', 'Dog', 'Horse',
'Brittle stars', 'Centipede']})
pq.write_table(table, f'tos://{bucket}/data.parquet', filesystem=fs)
# 读取 Parquet 数据集
dataset = ds.dataset(f'tos://{bucket}/data.parquet', filesystem=fs)
# 输出数据集信息
print(f'Rows: {dataset.count_rows()}')
print(dataset.schema)
# 转换为 Pandas DataFrame
df = dataset.to_table().to_pandas()
print(df.head())
与 PyTorch 集成
PyTorch可以在数据集加载和 Checkpoint 读写等场景集成 tosfsspec。
代码示例
import fsspec
import torch
import tosfsspec
from torchdata.datapipes.iter import FSSpecFileLister, FSSpecFileOpener
import os
import tos
bucket = 'your-bucket'
# 从环境变量获取 AKSK 信息
key = os.getenv('TOS_ACCESS_KEY')
secret = os.getenv('TOS_SECRET_KEY')
# 数据集加载
fsspec.register_implementation('tos-new', 'tosfsspec.TosFileSystem', )
storage_options = {
'key': key,
'secret': secret,
'region': 'cn-beijing'
}
file_lister = FSSpecFileLister(root=f'tos-new://{bucket}/dataset/', **storage_options)
iterable_dataset = FSSpecFileOpener(file_lister, mode='rb', **storage_options)
for _, item in iterable_dataset:
data = item.read()
# Checkpoint 读写
fs = tosfsspec.TosFileSystem(
key=key,
secret=secret,
region='cn-beijing'
)
model = torch.nn.Linear(10, 10)
with fs.open(f'tos://{bucket}/models/model.pt', 'wb') as f:
torch.save(model.state_dict(), f)
with fs.open(f'tos://{bucket}/models/model.pt', 'rb') as f:
state_dict = torch.load(f)
# print(state_dict)
model.load_state_dict(state_dict)
最近更新时间：2026.03.27 18:00:01
这个页面对您有帮助吗？
有用
有用
无用
无用