Simpan Dataframe ke csv langsung ke s3 Python

Saya memiliki DataFrame panda yang ingin saya unggah ke file CSV baru. Masalahnya adalah saya tidak ingin menyimpan file secara lokal sebelum mentransfernya ke s3. Apakah ada metode seperti to_csv untuk menulis dataframe ke s3 secara langsung? Saya menggunakan boto3.
Inilah yang saya miliki sejauh ini:

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

Kamu bisa memakai:

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

Anda dapat langsung menggunakan jalur S3. Saya menggunakan 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

Catatan Rilis:

Penanganan File S3

panda sekarang menggunakan s3fs untuk menangani koneksi S3. Ini seharusnya tidak merusak kode apa pun. Namun, karena s3fs bukanlah dependensi yang diperlukan, Anda perlu menginstalnya secara terpisah, seperti boto di versi panda sebelumnya. GH11915 .

Saya suka s3fs yang memungkinkan Anda menggunakan s3 (hampir) seperti sistem file lokal.

Kamu bisa melakukan ini:

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)

s3fshanya mendukung rbdan wbmode membuka file, itulah mengapa saya melakukan bytes_to_writehal ini .

Ini adalah jawaban yang lebih mutakhir:

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)

Masalah dengan StringIO adalah ia akan menggerogoti memori Anda. Dengan metode ini, Anda mengalirkan file ke s3, daripada mengubahnya menjadi string, lalu menulisnya menjadi s3. Memegang bingkai data panda dan salinan stringnya di memori tampaknya sangat tidak efisien.

Jika Anda bekerja dalam ec2 instan, Anda dapat memberikan peran IAM untuk mengaktifkan penulisannya ke s3, sehingga Anda tidak perlu memberikan kredensial secara langsung. Namun, Anda juga dapat terhubung ke bucket dengan meneruskan kredensial ke S3FileSystem()fungsi tersebut. Lihat dokumentasinya: https://s3fs.readthedocs.io/en/latest/

Jika Anda lolos Nonesebagai argumen pertama keto_csv() , data akan dikembalikan sebagai string. Dari sana, langkah mudah untuk mengunggahnya ke S3 dalam sekali jalan.

Itu juga harus memungkinkan untuk mengirimkan StringIOobjek ke to_csv(), tetapi menggunakan string akan lebih mudah.

Saya menemukan ini dapat dilakukan dengan menggunakan clientjuga dan tidak hanya 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')

Anda juga dapat menggunakan AWS Data Wrangler :

import awswrangler

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

Perhatikan bahwa itu akan dibagi menjadi beberapa bagian karena mengunggahnya secara paralel.

karena Anda menggunakan boto3.client(), coba:

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

Saya menemukan solusi yang sangat sederhana yang tampaknya berhasil:

s3 = boto3.client("s3")

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

Semoga membantu!

Saya membaca csv dengan dua kolom dari ember s3, dan isi file csv saya masukkan ke dalam pandas dataframe.

Contoh:

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