s3 selectでs3からデータを取得する - 闘うITエンジニアの覚え書き

csv データのパースは csv.reader を使用。(pandas 等は使用しない)

s3_select_csv.py

import argparse
import boto3
import json
import time
import re
import csv 

BUCKET_NAME = 'バケット名'
OBJECT_KEY = 'test-s3-select.csv'


def init(count):
    """ 
    テストデータ作成.
    """
    print(f"データ作成 件数:{count}")
    header = 'no,data1,data2,data3,data4,data5'
    items = '\n'.join([f'{i},data{i},data{i+1},data{i+2},data{i+3},data{i+4}' for i in range(count)])
    bytes_items = f'{header}\n{items}'.encode('utf-8')
    client = boto3.client('s3')
    client.put_object(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        Body=bytes_items
    )   


def normal_filter():
    """ 
    APIで全データ取得してロジックでフィルタを書ける場合.
    """
    client = boto3.client('s3')
    stime = time.time()
    response = client.get_object(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY
    )   
    target = []
    if 'Body' in response:
        body = response['Body'].read()
        records_text = body.decode('utf-8')
        items = [x for x in csv.reader(records_text.strip().splitlines())]
        target = list(filter(lambda x: x[1] == 'data100', items))
    etime = time.time()
    print('処理時間(通常): {}'.format(etime - stime))
    print(target)


def s3select_filter():
    """ 
    S3 select で条件指定してデータ取得する場合.
    """
    client = boto3.client('s3')
    stime = time.time()
    response = client.select_object_content(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        ExpressionType='SQL',
        Expression="SELECT * FROM S3Object WHERE data1 = 'data100'",
        InputSerialization={'CompressionType': 'NONE', 'CSV': {'FileHeaderInfo': 'USE', 'RecordDelimiter': '\n', 'FieldDelimiter': ','}},
        OutputSerialization={'CSV': {'RecordDelimiter': '\n', 'FieldDelimiter': ','}}
    )   
    target = []
    if 'Payload' in response:
        # botocore.eventstream.EventStream からデータを取り出し
        for event_stream in response['Payload']:
            if 'Records' in event_stream and 'Payload' in event_stream['Records']:
                records_text = event_stream['Records']['Payload'].decode('utf-8')
                items = [x for x in csv.reader(records_text.strip().splitlines())]
                target = target + items
    etime = time.time()
    print('処理時間(s3-select): {}'.format(etime - stime))
    print(target)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='s3 selectのパフォーマンス検証')
    parser.add_argument('--init', action='store_true', help='初期処理(データ作成)時に指定')
    parser.add_argument('--count', type=int, default=10, help='データ作成件数')
    args = parser.parse_args()
    if (args.init):
        init(args.count)
    else:
        print("")
        normal_filter()
        s3select_filter()

python3 s3_select_csv --init --count 10000

python3 s3_select_csv

処理時間(通常): 0.909980058670044
[['100', 'data100', 'data101', 'data102', 'data103', 'data104']]
処理時間(s3-select): 0.396359920501709
[['100', 'data100', 'data101', 'data102', 'data103', 'data104']]

s3_select_json.py

import argparse
import boto3
import json
import time
import re

BUCKET_NAME = 'バケット名'
OBJECT_KEY = 'test-s3-select.json'


def init(count):
    print(f"データ作成 件数:{count}")
    bytes_items = json.dumps({
            'items':[{'no': i, 'data1': f'data{i}', 'data2': f'data{i+1}',
            'data3': f'data{i+2}', 'data4': f'data{i+3}', 'data5': f'data{i+4}'
        } for i in range(count)]}).encode('utf-8')
    client = boto3.client('s3')
    client.put_object(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        Body=bytes_items
    )   


def normal_filter():
    """ 
    APIで全データ取得してロジックでフィルタを書ける場合.
    """
    client = boto3.client('s3')
    stime = time.time()
    response = client.get_object(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY
    )   
    target = []
    if 'Body' in response:
        body = response['Body'].read()
        text = body.decode('utf-8')
        items = json.loads(text)
        items = items['items']
        target = list(filter(lambda x: x['data1'] == 'data100', items))
    etime = time.time()
    print('処理時間(通常): {}'.format(etime - stime))
    print(target)


def s3select_filter():
    """ 
    S3 select で条件指定してデータ取得する場合.
    """
    client = boto3.client('s3')
    stime = time.time()
    response = client.select_object_content(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        ExpressionType='SQL',
        Expression="SELECT d.* FROM S3Object[*].items[*] d WHERE d.data1 = 'data100'",
        InputSerialization={'CompressionType': 'NONE', 'JSON': {'Type': 'DOCUMENT'}},
        OutputSerialization={'JSON': {'RecordDelimiter': ','}}  # json.loadsしやすいように区切りはカンマにしておく
    )   
    target = []
    if 'Payload' in response:
        # botocore.eventstream.EventStream からデータを取り出し
        for event_stream in response['Payload']:
            if 'Records' in event_stream and 'Payload' in event_stream['Records']:
                records_text = event_stream['Records']['Payload'].decode('utf-8')
                records_text = "[" + re.sub(",$", "", records_text) + "]"  # 最後尾のカンマを除去
                rec = json.loads(records_text)
                target = target + rec 
    etime = time.time()
    print('処理時間(s3-select): {}'.format(etime - stime))
    print(target)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='s3 selectのパフォーマンス検証')
    parser.add_argument('--init', action='store_true', help='初期処理(データ作成)時に指定')
    parser.add_argument('--count', type=int, default=10, help='データ作成件数')
    args = parser.parse_args()
    if (args.init):
        init(args.count)
    else:
        print("")
        normal_filter()
        s3select_filter()

python3 s3_select_json.py --init --count 10000

python3 s3_select_json.py

処理時間(通常): 1.464644193649292
[{'no': 100, 'data1': 'data100', 'data2': 'data101', 'data3': 'data102', 'data4': 'data103', 'data5': 'data104'}]
処理時間(s3-select): 0.32223010063171387
[{'no': 100, 'data1': 'data100', 'data2': 'data101', 'data3': 'data102', 'data4': 'data103', 'data5': 'data104'}]