Salve Dataframe em csv diretamente no s3 Python

125

Eu tenho um DataFrame do pandas que desejo enviar para um novo arquivo CSV. O problema é que não quero salvar o arquivo localmente antes de transferi-lo para o s3. Existe algum método como to_csv para gravar o dataframe em s3 diretamente? Estou usando o boto3.
Aqui está o que tenho até agora:

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
user2494275
fonte
3
df.to_csv('s3://mybucket/dfs/somedf.csv'). stackoverflow.com/a/56275519/908886 para mais informações.
Peter Berg

Respostas:

158

Você pode usar:

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())
Stefan
fonte
9
Se este for um arquivo grande, o que isso causa na memória ...?
citynorman
2
Se o arquivo for maior que a RAM disponível, a ação irá falhar e irá exceto uma Exceção (não sei qual). Isso deve ser aceito como resposta
Eran Moshe
5
Recebi um TypeError: unicode argument expected, got 'str'erro ao usar StringIO. Usei BytesIOe funcionou perfeitamente bem. Observação: isso foi em Python 2.7
Abhishek Upadhyaya
1
o que é bucketobjeto? como você criou isso?
Charles Chow
1
bucketé onde você armazena objetos no S3. O código assume que você já criou o destino (pense: diretório) onde armazená-lo. Ver S3 docs
Stefan
65

Você pode usar diretamente o caminho S3. Estou usando o 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

Nota de lançamento:

Tratamento de arquivos S3

O pandas agora usa s3fs para lidar com conexões S3. Isso não deve quebrar nenhum código. No entanto, como o s3fs não é uma dependência necessária, você precisará instalá-lo separadamente, como o boto nas versões anteriores do pandas. GH11915 .

critério 17
fonte
7
esta é definitivamente a resposta mais fácil agora, ele usa s3fs nos bastidores, então você precisa adicioná-lo ao seu requirements.txt
JD D
1
Eu gosto de ser fácil, mas parece que não está funcionando, visto que continuo recebendo o seguinte erro NoCredentialsError: Unable to locate credentials. Alguma sugestão?
CathyQian
1
Posso confirmar que isso não funciona com pandas <= 0.23.4, então certifique-se de atualizar para pandas 0.24
Guido
1
Este é o erro que vejo quando tento usar o comando to_csv TypeError: write () o argumento 1 deve ser Unicode, não str
Raj
13
Estou usando o pandas 0.24.2 e o que recebo é NotImplementedError: Text mode not supported, use mode='wb' and manage bytes. alguma sugestão?
Binyamin Even
57

Eu gosto do s3fs, que permite que você use o s3 (quase) como um sistema de arquivos local.

Você consegue fazer isso:

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)

s3fssuporta apenas rbe wbmodos de abrir o arquivo, é por isso que fiz isso bytes_to_write.

michcio1234
fonte
Ótimo! Como posso obter o url do arquivo usando o mesmo módulo s3fs?
M.Zaman
Eu estava procurando a URL de onde posso baixar o arquivo escrito, de qualquer forma, consigo isso via S3FileSystem. Obrigado
M.Zaman,
é isso que eu uso; obrigado. Estou curioso para saber por que pd.read_csv (<s3path>) funciona como esperado, mas para escrever, temos que usar esta solução alternativa .. exceto no caso de eu estar escrevendo diretamente para o balde s3 em que meu jupyter está.
Renée
@ michcio1234 como posso fazer o mesmo no modo anexar? Preciso anexar os dados no csv existente em s3
j '
@j ' s3fsnão parece suportar o modo de acréscimo.
michcio1234
43

Esta é uma resposta mais atualizada:

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)

O problema com o StringIO é que ele vai consumir sua memória. Com este método, você está transmitindo o arquivo para s3, em vez de convertê-lo em string e, em seguida, gravá-lo em s3. Manter o dataframe do pandas e sua cópia da string na memória parece muito ineficiente.

Se você estiver trabalhando em um instante ec2, poderá atribuir a ele uma função IAM para permitir gravá-lo em s3, portanto, não é necessário passar credenciais diretamente. No entanto, você também pode se conectar a um intervalo passando credenciais para a S3FileSystem()função. Veja a documentação: https://s3fs.readthedocs.io/en/latest/

erncyp
fonte
Por alguma razão, quando fiz isso, todas as linhas foram puladas na saída CSV
kjmerf
Hmm. não tenho certeza por que isso aconteceria. talvez tente com outro pandas df para ver se você ainda pega o problema? Se a sua versão do pandas for compatível, tente a resposta de @amit-kushwaha, para onde você passa o url s3 diretamente to_csv(). parece uma implementação mais limpa.
erncyp
@erncyp Parece que estou recebendo o erro: botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied ... Eu até mesmo fiz o intervalo PUBLIC READ e adicionei as seguintes ações, sob minha conta específica de usuário IAM, na Política do "Action": [ "s3:PutObject", "s3:PutObjectAcl", "s3:GetObject", "s3:GetObjectAcl", "s3:DeleteObject" ]
intervalo
parece que você está sem as permissões? Certifique-se de anexar permissões de leitura e gravação S3 à função IAM que você está usando
erncyp
@erncyp Eu tenho a política de Acesso do Administrador anexada ao meu usuário IAM, então, em teoria, eu deveria ser capaz de ler / escrever muito bem ... Estranhamente, eu sou capaz de escrever muito bem quando uso a seguinte função que criei, usando outro usuário do StackOverflow conselho (ponto e vírgula fyi são fim de linha, pois não sei como formatar na seção de comentários):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());
ajoros
13

Se passar Nonecomo primeiro argumento to_csv()os dados serão devolvidos como string. A partir daí, é uma etapa fácil fazer o upload para o S3 de uma vez.

Também deve ser possível passar um StringIOobjeto para to_csv(), mas usar uma string será mais fácil.

mhawke
fonte
Será mais fácil de que maneira? Qual é a maneira correta de fazer isso?
Eran Moshe
@EranMoshe: de qualquer forma vai funcionar correctamente, mas, obviamente, é mais fácil de passar Nonepara to_csv()e usar a string retornada do que é criar um StringIOobjeto e, em seguida, ler novamente os dados para fora.
mhawke
Como um programador preguiçoso, foi o que fiz. E você quis mais fácil para o programador que escreve menos código:>
Eran Moshe
2

Você também pode usar o AWS Data Wrangler :

import awswrangler

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

Observe que ele será dividido em várias partes, uma vez que carrega em paralelo.

gabra
fonte
2

Descobri que isso pode ser feito usando clienttambém e não apenas 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')
Harry_pb
fonte
0

já que você está usando boto3.client(), tente:

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')
Jerrytim
fonte
-1

Encontrei uma solução muito simples que parece estar funcionando:

s3 = boto3.client("s3")

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

Espero que ajude !

Antoine Krajnc
fonte
-5

Eu li um csv com duas colunas do intervalo s3 e o conteúdo do arquivo csv eu coloquei no dataframe do pandas.

Exemplo:

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()
Jamir Josimar Huamán Campos
fonte
4
por favor, não publique apenas a solução, acrescente uma explicação sobre ela.
sjaustirni de
Existe alguma vantagem em fazer uma solução tão complexa (para um novato em Python)?
Javier López Tomás
1
Isso lê um arquivo de s3, a questão era como escrever um df para s3.
Damian Satterthwaite-Phillips