【Python3】S3のcsvやエクセルをpandasで読み込む

はじめに

AWSのS3にあるcsvファイルやエクセルファイルを,ダウンロードせずに直接読み込みたい!と思った

pandasに標準装備されている

参考: pandas DataFrameをs3から読む・s3に出力するメモ

import pandas as pd

df = pd.read_csv('s3://your-backet/your-file.csv')

で簡単に読み込めてしまう．
しかし，これは ~/.aws/credentials に

[default]
aws_access_key_id = hogehoge
aws_secret_access_key = hogehoge

と,defaultプロファイルに該当のAWSアカウントのS3にアクセスできるアクセスキーが記載されている場合のみ．

実際には ~/.aws/credentials はこんな感じになることが多いと思います

[default]
aws_access_key_id = hogehoge
aws_secret_access_key = hogehoge

[sample_read_s3]
aws_access_key_id = hogehogee
aws_secret_access_key = hogehogeee

RW_S3 package

そのため，プロファイル名を指定してS3のcsvやエクセルファイルにアクセスするpackageを作りました．

install

pip install RW_S3

でいけます

RW_S3モジュールにはread_s3とwrite_s3の二つのクラスを用意していて，

• read_s3 -> s3にあるcsvやエクセルファイルの読み込み
• write_s3 -> データフレームを直接s3にアップロード

といった感じになっています．

usage

読み込み

from RW_S3 import read_s3
s3 = read_s3("sample_read_s3")
df = s3.read_csv(bucket = "your-backet", key = "your-file.csv")

書き出し

from RW_S3 import write_s3
w_s3.to_csv(df, bucket = "your-backet", key = "your-file.csv", index=False)

chords

• read_s3.py

import pandas as pd
import boto3
from boto3.session import Session
import re

class read_s3(object):
    def __init__(self, s3_profile):
        session = Session(profile_name=s3_profile)  # s3にアクセスするためのプロファイルを指定
        self.__s3 = session.client('s3')
        self.object_info = {}

    def ls(self, bucket, key="", last_modified_time=False):
        """
        bucketとkeyを指定したらその配下のオブジェクトを全てリストで返す
        """
        list_objects = self.__s3.list_objects(
            Bucket=bucket,
            Prefix=key
        )
        if "Contents" in list_objects:
            if last_modified_time == False:
                return [content["Key"] for content in list_objects["Contents"]]
            else:
                return {content["Key"]: content["LastModified"].strftime('%Y-%m-%d_%H:%M:%S')
                        for content in list_objects["Contents"]}

    def read_csv(self, bucket, key, encoding="utf_8", sep=',', header=0, index_col=None, usecols=None, na_values=None, nrows=None, skiprows=0):
        """
        csvの読み込み(pandasのread_csvとほぼ同じ)
            bucket: s3のバケット名
            key: バケット内のkey名
        """
        read_file = self.__s3.get_object(Bucket=bucket, Key=key)
        df = pd.read_csv(read_file['Body'], encoding=encoding, sep=sep, header=header,
                         index_col=index_col, usecols=usecols, na_values=na_values, nrows=nrows, skiprows=skiprows)
        object_info_dic = self.ls(bucket=bucket, key=key, last_modified_time=True)
        print("object_info: %s" % object_info_dic)
        self.object_info.update(object_info_dic)
        return df

    def read_excel(self, bucket, key, encoding="utf_8", sheet_name=0, header=0, index_col=None, usecols=None, na_values=None, nrows=None, skiprows=0):
        """
        excelの読み込み(pandasのread_excelとほぼ同じ)
            bucket: s3のバケット名
            key: バケット内のkey名
        """
        read_file = self.__s3.get_object(Bucket=bucket, Key=key)
        df = pd.read_excel(read_file['Body'], encoding=encoding, sheet_name=sheet_name,
                           header=header, index_col=index_col, usecols=usecols, na_values=na_values, nrows=nrows, skiprows=skiprows)
        object_info_dic = self.ls(bucket=bucket, key=key, last_modified_time=True)
        print("object_info: %s" % object_info_dic)
        self.object_info.update(object_info_dic)
        return df

    def read_table(self, bucket, key, encoding="utf_8", sep="\t", header=0, index_col=None, usecols=None, na_values=None, nrows=None):
        """
        ※※※※ pandasではread_tableは非推奨扱いです ※※※※
        ※※※※ read_csv(sep='\t')を用いてください ※※※※
        csv, tsv, excel等の読み込み(pandasのread_tableとほぼ同じ)
            bucket: s3のバケット名
            key: バケット内のkey名
        """
        read_file = self.__s3.get_object(Bucket=bucket, Key=key)
        df = pd.read_table(read_file['Body'], encoding=encoding, header=header, sep=sep,
                         index_col=index_col, usecols=usecols, na_values=na_values, nrows=nrows)
        object_info_dic = self.ls(bucket=bucket, key=key, last_modified_time=True)
        print("object_info: %s" % object_info_dic)
        self.object_info.update(object_info_dic)
        return df

• wrtie_s3.py

import re
import pandas as pd
import boto3
from boto3.session import Session

class write_s3():
    def __init__(self, s3_profile='put_s3_lambda'):
        session = Session(profile_name=s3_profile)  # s3にアクセスするためのプロファイルを指定
        self.__s3 = session.client('s3')
        
    def to_csv(self, df: pd.DataFrame, bucket, key, index=True, encoding="utf_8"):
        """
        変数dfをbucketバケットのkeyパスに書き出す
            * df: pandasのデータフレーム
            * bucket: アップロードしたいバケット名
            * key: アップロードしたバケット内のファイルパス
        ※ 存在しないバケットにもアップロードできます
        """
        bytes_to_write = df.to_csv(None, index=index, encoding=encoding).encode(encoding)
        self.__s3.put_object(ACL='private', Body=bytes_to_write, Bucket=bucket, Key=key)

まだREADMEや機能も充実していませんが，githubのv0.0.2ブランチで追加開発します．