Speichern Sie Dataframe in CSV direkt in S3 Python

Ich habe einen Pandas DataFrame, den ich in eine neue CSV-Datei hochladen möchte. Das Problem ist, dass ich die Datei nicht lokal speichern möchte, bevor ich sie auf s3 übertrage. Gibt es eine Methode wie to_csv, um den Datenrahmen direkt in s3 zu schreiben? Ich benutze boto3.
Folgendes habe ich bisher:

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

Sie können verwenden:

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

Sie können den S3-Pfad direkt verwenden. Ich benutze 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

Veröffentlichungshinweis:

S3 Dateiverwaltung

pandas verwendet jetzt s3fs für die Behandlung von S3-Verbindungen. Dies sollte keinen Code beschädigen. Da s3fs jedoch keine erforderliche Abhängigkeit ist, müssen Sie es separat installieren, wie Boto in früheren Versionen von Pandas. GH11915 .

Ich mag s3fs, mit dem Sie s3 (fast) wie ein lokales Dateisystem verwenden können.

Du kannst das:

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)

s3fsunterstützt nur rbund wbModi zum Öffnen der Datei, deshalb habe ich dieses bytes_to_writeZeug gemacht.

Dies ist eine aktuellere Antwort:

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)

Das Problem mit StringIO ist, dass es Ihr Gedächtnis auffrisst. Mit dieser Methode streamen Sie die Datei in s3, anstatt sie in einen String zu konvertieren, und schreiben sie dann in s3. Das Halten des Pandas-Datenrahmens und seiner Zeichenfolgenkopie im Speicher scheint sehr ineffizient zu sein.

Wenn Sie in einem ec2-Moment arbeiten, können Sie ihm eine IAM-Rolle zuweisen, damit er in s3 geschrieben werden kann. Sie müssen also keine Anmeldeinformationen direkt übergeben. Sie können jedoch auch eine Verbindung zu einem Bucket herstellen, indem Sie Anmeldeinformationen an die S3FileSystem()Funktion übergeben. Siehe Dokumentation: https://s3fs.readthedocs.io/en/latest/

Wenn Sie Noneals erstes Argument an to_csv()die Daten übergeben, werden diese als Zeichenfolge zurückgegeben. Von dort aus ist es ein einfacher Schritt, das auf einmal in S3 hochzuladen.

Es sollte auch möglich sein, ein StringIOObjekt an zu übergeben to_csv(), aber die Verwendung einer Zeichenfolge ist einfacher.

Sie können auch den AWS Data Wrangler verwenden :

import awswrangler

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

Beachten Sie, dass es in mehrere Teile aufgeteilt wird, da es parallel hochgeladen wird.

Ich fand, dass dies clientauch mit und nicht nur möglich ist 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')

boto3.client()Versuchen Sie, da Sie verwenden :

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

Ich habe eine sehr einfache Lösung gefunden, die zu funktionieren scheint:

s3 = boto3.client("s3")

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

Hoffentlich hilft das !

Ich habe eine CSV mit zwei Spalten aus Bucket S3 gelesen und den Inhalt der Datei CSV in Pandas Dataframe eingefügt.

Beispiel:

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