使用Minio与Dask时遇's3.ServiceResource'无'create_client'属性错误

我来帮你理清这个问题的根源和解决办法：

问题根源 #

你创建的s3变量是boto3的ServiceResource对象，而Dask依赖的s3fs库在处理storage_options里的session参数时，期望的是boto3的Session对象或者Client对象——这两者才具备s3fs需要的create_client方法，而高层封装的Resource对象并没有这个接口。

你之前用Resource遍历桶能成功，是因为Resource本身就封装了桶列表查询的逻辑，但s3fs需要更底层的客户端来处理文件系统级别的请求，所以才会触发这个错误。

解决办法 #

有两种简单的修改方式，任选其一即可：

方式1：直接使用boto3 Client对象

把你的boto3配置代码改成创建Client（而非Resource）：

import boto3, botocore, os
from botocore.client import Config

# 创建boto3 Client对象
s3_client = boto3.client(
    's3',
    endpoint_url='https://example.org',
    aws_access_key_id=key,
    aws_secret_access_key=secret,
    config=Config(signature_version='s3v4'),
    region_name='us-east-1'
)
os.environ['S3_USE_SIGV4'] = 'True'

然后在Dask代码里，把storage_options的参数改成client：

from minio import Minio
from minio.error import ResponseError
import io, sys, dask, s3fs
import pandas as pd
import dask.bag as db
import json

# 使用client参数传递配置
df = db.read_text('s3://mybucket/prefa/prefb/*.jsonld', storage_options={"client": s3_client}).map(json.loads)

方式2：使用boto3 Session对象

如果你想保留Session级别的配置，可以直接创建Session并传给s3fs：

import boto3, botocore, os
from botocore.client import Config
from botocore.session import Session

# 创建Session对象
session = Session()
# 基于Session创建Client（也可以直接把session传给storage_options）
s3_client = session.create_client(
    's3',
    endpoint_url='https://example.org',
    aws_access_key_id=key,
    aws_secret_access_key=secret,
    config=Config(signature_version='s3v4'),
    region_name='us-east-1'
)
os.environ['S3_USE_SIGV4'] = 'True'

Dask代码可以选两种写法之一：

# 写法1：传递client
df = db.read_text('s3://mybucket/prefa/prefb/*.jsonld', storage_options={"client": s3_client}).map(json.loads)

# 写法2：直接传递session
df = db.read_text('s3://mybucket/prefa/prefb/*.jsonld', storage_options={"session": session}).map(json.loads)

验证修改是否有效 #

你可以先测试新创建的Client/Session是否能正常访问Minio：

# 测试Client
response = s3_client.list_buckets()
for bucket in response['Buckets']:
    print(bucket['Name'])

# 测试Session（如果用Session的话）
test_client = session.create_client('s3')
response = test_client.list_buckets()
for bucket in response['Buckets']:
    print(bucket['Name'])

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