将 Dataframe 保存到 csv 直接保存到 s3 Python
Posted
技术标签:
【中文标题】将 Dataframe 保存到 csv 直接保存到 s3 Python【英文标题】:Save Dataframe to csv directly to s3 Python 【发布时间】:2016-11-04 09:00:12 【问题描述】:我有一个要上传到新 CSV 文件的 pandas DataFrame。问题是我不想在将文件传输到 s3 之前将其保存在本地。有没有像 to_csv 这样的方法直接将数据帧写入 s3?我正在使用 boto3。 这是我目前所拥有的:
import boto3
s3 = boto3.client('s3', aws_access_key_id='key', aws_secret_access_key='secret_key')
read_file = s3.get_object(Bucket, Key)
df = pd.read_csv(read_file['Body'])
# Make alterations to DataFrame
# Then export DataFrame to CSV through direct transfer to s3
【问题讨论】:
df.to_csv('s3://mybucket/dfs/somedf.csv')
。 ***.com/a/56275519/908886 了解更多信息。
【参考方案1】:
你可以使用:
from io import StringIO # python3; python2: BytesIO
import boto3
bucket = 'my_bucket_name' # already created on S3
csv_buffer = StringIO()
df.to_csv(csv_buffer)
s3_resource = boto3.resource('s3')
s3_resource.Object(bucket, 'df.csv').put(Body=csv_buffer.getvalue())
【讨论】:
如果这是一个大文件,这对内存有什么影响...? 如果文件更大,那么您可用的 RAM 将失败,并且会出现异常(不知道是哪一个)。这应该被接受为答案 我在使用StringIO
时遇到了TypeError: unicode argument expected, got 'str'
错误。我使用了BytesIO
,它工作得非常好。 注意:这是在 Python 2.7 中
什么是bucket
对象?你是怎么做到的?
bucket
是您在 S3 上存储对象的位置。该代码假定您已经创建了存储它的目标(想想:目录)。见S3 docs【参考方案2】:
您可以直接使用 S3 路径。我正在使用Pandas 0.24.1
In [1]: import pandas as pd
In [2]: df = pd.DataFrame( [ [1, 1, 1], [2, 2, 2] ], columns=['a', 'b', 'c'])
In [3]: df
Out[3]:
a b c
0 1 1 1
1 2 2 2
In [4]: df.to_csv('s3://experimental/playground/temp_csv/dummy.csv', index=False)
In [5]: pd.__version__
Out[5]: '0.24.1'
In [6]: new_df = pd.read_csv('s3://experimental/playground/temp_csv/dummy.csv')
In [7]: new_df
Out[7]:
a b c
0 1 1 1
1 2 2 2
Release Note:
S3 文件处理
pandas 现在使用 s3fs 来处理 S3 连接。这不应该破坏任何代码。但是,由于 s3fs 不是必需的依赖项,因此您需要单独安装它,就像以前版本的 pandas 中的 boto 一样。 GH11915.
【讨论】:
这绝对是现在最简单的答案,它在后台使用 s3fs,因此您需要将其添加到您的 requirements.txt 中 我喜欢它很简单,但它似乎并没有真正起作用,因为我不断收到以下错误NoCredentialsError: Unable to locate credentials
。有什么建议吗?
我可以确认这不适用于 pandas
这是我在尝试使用 to_csv 命令时看到的错误 TypeError: write() argument 1 must be unicode, not str
我使用的是 pandas 0.24.2,我得到的是NotImplementedError: Text mode not supported, use mode='wb' and manage bytes
。有什么建议吗?【参考方案3】:
我喜欢s3fs,它可以让您(几乎)像使用本地文件系统一样使用 s3。
你可以这样做:
import s3fs
bytes_to_write = df.to_csv(None).encode()
fs = s3fs.S3FileSystem(key=key, secret=secret)
with fs.open('s3://bucket/path/to/file.csv', 'wb') as f:
f.write(bytes_to_write)
s3fs
仅支持rb
和wb
模式打开文件,这就是我这样做bytes_to_write
的原因。
【讨论】:
太棒了!如何使用相同的 s3fs 模块获取文件 url? 我正在寻找可以下载书面文件的 URL,无论如何我是通过 S3FileSystem 获得的。谢谢 这是我用的;谢谢。我很好奇为什么 pd.read_csv(s3fs
似乎不支持追加模式。【参考方案4】:
这是一个更新的答案:
import s3fs
s3 = s3fs.S3FileSystem(anon=False)
# Use 'w' for py3, 'wb' for py2
with s3.open('<bucket-name>/<filename>.csv','w') as f:
df.to_csv(f)
StringIO 的问题是它会侵蚀你的记忆。使用此方法,您将文件流式传输到 s3,而不是将其转换为字符串,然后将其写入 s3。在内存中保存 pandas 数据帧及其字符串副本似乎非常低效。
如果您在 ec2 即时工作,您可以给它一个 IAM 角色以允许将其写入 s3,因此您不需要直接传递凭证。但是,您也可以通过将凭据传递给 S3FileSystem()
函数来连接到存储桶。见文档:https://s3fs.readthedocs.io/en/latest/
【讨论】:
由于某种原因,当我这样做时,输出 CSV 中的每一行都被跳过了 嗯。不知道为什么会这样。也许尝试使用另一个 pandas df 看看您是否仍然遇到问题?如果您的 pandas 版本支持它,请尝试 @amit-kushwaha 的答案,将 s3 url 直接传递给to_csv()
。似乎是一个更干净的实现。
@erncyp 我似乎到了那里错误:botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied
...我什至已将存储桶设为 PUBLIC READ,并且在我的特定帐户 IAM 用户下,在存储桶中添加了以下操作政策:"Action": [ "s3:PutObject", "s3:PutObjectAcl", "s3:GetObject", "s3:GetObjectAcl", "s3:DeleteObject" ]
您好像没有权限?确保将 S3 读写权限附加到您正在使用的 IAM 角色
@erncyp 我的 IAM 用户附加了 AdministratorAccess 策略,所以理论上我应该能够很好地读/写......奇怪的是,当我使用以下内容时,我能够写得很好我使用另一个 *** 用户的建议制作的函数(fyi 分号是行尾,因为我不知道如何在评论部分格式化):def send_to_bucket(df, fn_out, bucketname): csv_buffer = StringIO(); df.to_csv(csv_buffer); s3_resource = boto3.resource('s3'); s3_resource.Object(bucketname, fn_out).put(Body=csv_buffer.getvalue());
【参考方案5】:
如果您将None
作为第一个参数传递给to_csv()
,则数据将作为字符串返回。从那里可以轻松地将其一次性上传到 S3。
也应该可以将StringIO
对象传递给to_csv()
,但使用字符串会更容易。
【讨论】:
哪种方式会更容易?正确的做法是什么? @EranMoshe:任何一种方式都可以正常工作,但显然将None
传递给to_csv()
并使用返回的字符串比创建StringIO
对象然后读取数据更容易退出。
作为一个懒惰的程序员,我就是这么做的。对于编写更少代码的程序员来说,您的意思是更容易:>【参考方案6】:
您也可以使用AWS Data Wrangler:
import awswrangler as wr
wr.s3.to_csv(
df=df,
path="s3://...",
)
请注意,它会为您处理分段上传以加快上传速度。
【讨论】:
【参考方案7】:我发现这也可以使用client
来完成,而不仅仅是resource
。
from io import StringIO
import boto3
s3 = boto3.client("s3",\
region_name=region_name,\
aws_access_key_id=aws_access_key_id,\
aws_secret_access_key=aws_secret_access_key)
csv_buf = StringIO()
df.to_csv(csv_buf, header=True, index=False)
csv_buf.seek(0)
s3.put_object(Bucket=bucket, Body=csv_buf.getvalue(), Key='path/test.csv')
【讨论】:
【参考方案8】:我使用AWS Data Wrangler。例如:
import awswrangler as wr
import pandas as pd
# read a local dataframe
df = pd.read_parquet('my_local_file.gz')
# upload to S3 bucket
wr.s3.to_parquet(df=df, path='s3://mys3bucket/file_name.gz')
这同样适用于 csv 文件。使用带有正确文件扩展名的read_csv
和to_csv
,而不是read_parquet
和to_parquet
。
【讨论】:
【参考方案9】:由于您使用的是boto3.client()
,请尝试:
import boto3
from io import StringIO #python3
s3 = boto3.client('s3', aws_access_key_id='key', aws_secret_access_key='secret_key')
def copy_to_s3(client, df, bucket, filepath):
csv_buf = StringIO()
df.to_csv(csv_buf, header=True, index=False)
csv_buf.seek(0)
client.put_object(Bucket=bucket, Body=csv_buf.getvalue(), Key=filepath)
print(f'Copy df.shape[0] rows to S3 Bucket bucket at filepath, Done!')
copy_to_s3(client=s3, df=df_to_upload, bucket='abc', filepath='def/test.csv')
【讨论】:
【参考方案10】:你可以使用
熊猫 boto3 s3fs(版本≤0.4)我在路径中使用to_csv
和s3://
和storage_options
key = "folder/file.csv"
df.to_csv(
f"s3://YOUR_S3_BUCKET/key",
index=False,
storage_options=
"key": AWS_ACCESS_KEY_ID,
"secret": AWS_SECRET_ACCESS_KEY,
"token": AWS_SESSION_TOKEN,
,
【讨论】:
【参考方案11】:我找到了一个似乎可行的非常简单的解决方案:
s3 = boto3.client("s3")
s3.put_object(
Body=open("filename.csv").read(),
Bucket="your-bucket",
Key="your-key"
)
希望有帮助!
【讨论】:
【参考方案12】:我从存储桶 s3 中读取了一个包含两列的 csv,以及我放入 pandas 数据框中的文件 csv 的内容。
例子:
config.json
"credential":
"access_key":"xxxxxx",
"secret_key":"xxxxxx"
,
"s3":
"bucket":"mybucket",
"key":"csv/user.csv"
cls_config.json
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import json
class cls_config(object):
def __init__(self,filename):
self.filename = filename
def getConfig(self):
fileName = os.path.join(os.path.dirname(__file__), self.filename)
with open(fileName) as f:
config = json.load(f)
return config
cls_pandas.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pandas as pd
import io
class cls_pandas(object):
def __init__(self):
pass
def read(self,stream):
df = pd.read_csv(io.StringIO(stream), sep = ",")
return df
cls_s3.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import boto3
import json
class cls_s3(object):
def __init__(self,access_key,secret_key):
self.s3 = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)
def getObject(self,bucket,key):
read_file = self.s3.get_object(Bucket=bucket, Key=key)
body = read_file['Body'].read().decode('utf-8')
return body
test.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from cls_config import *
from cls_s3 import *
from cls_pandas import *
class test(object):
def __init__(self):
self.conf = cls_config('config.json')
def process(self):
conf = self.conf.getConfig()
bucket = conf['s3']['bucket']
key = conf['s3']['key']
access_key = conf['credential']['access_key']
secret_key = conf['credential']['secret_key']
s3 = cls_s3(access_key,secret_key)
ob = s3.getObject(bucket,key)
pa = cls_pandas()
df = pa.read(ob)
print df
if __name__ == '__main__':
test = test()
test.process()
【讨论】:
请不要只发布解决方案,还要添加说明。 制作这样一个复杂的(对于 Python 新手)解决方案有什么好处吗? 这会从 s3 读取文件,问题是如何将 df 写入 s3。以上是关于将 Dataframe 保存到 csv 直接保存到 s3 Python的主要内容,如果未能解决你的问题,请参考以下文章
尝试在 DBFS 上将 Spark Dataframe 以 CSV 格式保存到磁盘