python - 从 Pandas 上传到 S3 时如何添加标签？

Question

Pandas 允许您将 AWS S3 路径直接传递给 .to_csv() 和 .to_parquet()。有一个用于传递 S3 特定参数的 storage_options 参数。

我想调用 .to_csv('s3://bucket/key.csv', storage_options=something) 并指定 S3 对象标签以应用于上传的对象，如 东西。我已经阅读了文档，但我不知道该怎么做，

The pandas docs 没有列出 storage_options 的可能值，它们只是指向 fsspec。看起来 pandas 调用 fsspec，fsspec 调用 s3fs，s3fs 调用 aiobotocore，aiobotocore 调用 botocore，并且可能调用 s3transfer。我怎样才能将 S3 标签参数一直传递到这个兔子洞里？

MWE

import pandas as pd
import boto3

bucket = 'mybucket' # change for your bucket
key = 'test/pandas/tags.csv'
tags = {'mytag': 'x'}

df = pd.DataFrame([{'a': 1}])
df.to_csv(f"s3://{bucket}/{key}") # try without any tags first
df.to_csv(f"s3://{bucket}/{key}", storage_options={'tags': tags})

resp = boto3.client('s3').get_object_tagging(Bucket=bucket, Key=key)
actual_tags = {t['Key']: t['Value'] for t in resp.get('TagSet', [])}
assert actual_tags == tags

预期行为

断言通过。 S3 对象具有标签 mytag:x

实际行为

第二个 .to_csv() 行失败。即它可以在没有标签的情况下工作。标签是导致失败的原因。

Traceback (most recent call last):
  File "upld.py", line 9, in <module>
    df.to_csv(f"s3://{bucket}/{key}", storage_options={'tags': tags})
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/pandas/core/generic.py", line 3463, in to_csv
    return DataFrameRenderer(formatter).to_csv(
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/pandas/io/formats/format.py", line 1105, in to_csv
    csv_formatter.save()
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/pandas/io/formats/csvs.py", line 237, in save
    with get_handle(
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/pandas/io/common.py", line 608, in get_handle
    ioargs = _get_filepath_or_buffer(
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/pandas/io/common.py", line 357, in _get_filepath_or_buffer
    file_obj = fsspec.open(
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/core.py", line 456, in open
    return open_files(
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/core.py", line 299, in open_files
    [fs.makedirs(parent, exist_ok=True) for parent in parents]
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/core.py", line 299, in <listcomp>
    [fs.makedirs(parent, exist_ok=True) for parent in parents]
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/asyn.py", line 91, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/asyn.py", line 71, in sync
    raise return_result
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/fsspec/asyn.py", line 25, in _runner
    result[0] = await coro
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/s3fs/core.py", line 746, in _makedirs
    await self._mkdir(path, create_parents=True)
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/s3fs/core.py", line 731, in _mkdir
    await self._call_s3("create_bucket", **params)
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/s3fs/core.py", line 252, in _call_s3
    await self.set_session()
  File "/home/ec2-user/.pyenv/versions/3.8.11/lib/python3.8/site-packages/s3fs/core.py", line 395, in set_session
    self.session = aiobotocore.session.AioSession(**self.kwargs)
TypeError: __init__() got an unexpected keyword argument 'tags'

看起来这些参数正在传递给 aiobotocore session 实例化，而不是来自 aiobotocore 的实际 S3 put_object API 调用。这让我觉得这是不可能的。

备选方案

我应该尝试:

storage_options={
    'tags': {
        'k': 'v'
    }
}

或

storage_options={
    'tags': [
        {'Key': 'k', 'Value': 'v'}
    ]
}

当然，我可以不带标签上传，然后将标签添加为单独的 boto 调用。这不是原子的，并且成本是原来的两倍(对于小文件)。如果有办法从上传中取回版本 ID，那将消除一些并发问题(并发写入)。

Answer 1

所以我花了一些时间在这方面进行挖掘。我在这里可能是错的，但我认为这是不可能的。这就是我相信的原因:

如果路径是不是以 http ( see here ) 开头的 url，

storage_options 将传递给 fsspec。然后这些选项通过 fsspec 作为 kwargs 传递给 s3fs.S3Filesystem。然后 kwargs 死胡同你的错误消息中的函数。

(这是我可能出错的地方!)然后 S3FileSystem 调用 _put_file 来编写您的 csv。此函数不使用 self.kwargs，但采用不会被 pandas 传递的函数级 kwargs。

因此，我认为在 pandas 中不能通过 to_X 访问标签。但是，值得在 Pandas/fsspec github 上提出问题以获取更多信息。