Zapisz Dataframe do csv bezpośrednio do s3 Python

Mam pandy DataFrame, które chcę przesłać do nowego pliku CSV. Problem w tym, że nie chcę zapisywać pliku lokalnie przed przesłaniem go do s3. Czy istnieje metoda, taka jak to_csv, do bezpośredniego zapisu ramki danych do s3? Używam boto3.
Oto, co mam do tej pory:

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

Możesz użyć:

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())

Możesz bezpośrednio użyć ścieżki S3. Używam Pandy 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

Informacje o wersji:

Obsługa plików S3

pandy używają teraz s3fs do obsługi połączeń S3. To nie powinno złamać żadnego kodu. Jednak ponieważ s3fs nie jest wymaganą zależnością, będziesz musiał zainstalować go osobno, tak jak boto we wcześniejszych wersjach pand. GH11915 .

Lubię s3fs, który pozwala ci używać s3 (prawie) jak lokalnego systemu plików.

Możesz to zrobić:

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)

s3fsobsługuje tylko rbi wbtryby otwierania pliku, dlatego zrobiłem to bytes_to_write.

Oto bardziej aktualna odpowiedź:

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)

Problem z StringIO polega na tym, że zżera twoją pamięć. Dzięki tej metodzie przesyłasz plik do s3, zamiast konwertować go na łańcuch, a następnie zapisywać do s3. Trzymanie ramki danych pandy i jej kopii w pamięci wydaje się bardzo nieefektywne.

Jeśli pracujesz w trybie ec2 natychmiast, możesz nadać mu rolę IAM, aby umożliwić zapisywanie go do s3, dzięki czemu nie musisz bezpośrednio przekazywać poświadczeń. Możesz jednak również połączyć się z zasobnikiem, przekazując dane uwierzytelniające do S3FileSystem()funkcji. Zobacz dokumentację: https://s3fs.readthedocs.io/en/latest/

Jeśli przekażesz Nonejako pierwszy argument, to_csv()dane zostaną zwrócone jako ciąg. Stamtąd łatwo jest przesłać to do S3 za jednym razem.

Powinno być również możliwe przekazanie StringIOobiektu to_csv(), ale użycie łańcucha będzie łatwiejsze.

Odkryłem, że można to zrobić clientrównież, a nie tylko 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')

Możesz również skorzystać z AWS Data Wrangler :

import awswrangler

session = awswrangler.Session()
session.pandas.to_csv(
    dataframe=df,
    path="s3://...",
)

Zauważ, że podzieli się na kilka części, ponieważ przesyła go równolegle.

ponieważ używasz boto3.client(), spróbuj:

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')

Znalazłem bardzo proste rozwiązanie, które wydaje się działać:

s3 = boto3.client("s3")

s3.put_object(
    Body=open("filename.csv").read(),
    Bucket="your-bucket",
    Key="your-key"
)

Mam nadzieję, że to pomoże!

Czytałem csv z dwiema kolumnami z wiadra s3, a zawartość pliku csv umieściłem w pandas dataframe.

Przykład:

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()