Amazon S3 からストリーミングデータをロードする

ディレクトリを作成します。このサンプルでは、名前 s3-to-opensearch を使用します。

sample.py という名前のディレクトリ内にファイルを作成します。

import boto3
import re
import requests
from requests_aws4auth import AWS4Auth

region = '' # e.g. us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-s3-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype

headers = { "Content-Type": "application/json" }

s3 = boto3.client('s3')

# Regular expressions used to parse some simple log lines
ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)')
time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]')
message_pattern = re.compile('\"(.+)\"')

# Lambda execution starts here
def handler(event, context):
    for record in event['Records']:

        # Get the bucket name and key for the new file
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Get, read, and split the file into lines
        obj = s3.get_object(Bucket=bucket, Key=key)
        body = obj['Body'].read()
        lines = body.splitlines()

        # Match the regular expressions to each line and index the JSON
        for line in lines:
            line = line.decode("utf-8")
            ip = ip_pattern.search(line).group(1)
            timestamp = time_pattern.search(line).group(1)
            message = message_pattern.search(line).group(1)

            document = { "ip": ip, "timestamp": timestamp, "message": message }
            r = requests.post(url, auth=awsauth, json=document, headers=headers)

region と host の変数を編集します。

まだ持っていない場合、pip をインストールしてから、依存関係を新しい package ディレクトリにインストールします。

cd s3-to-opensearch

pip install --target ./package requests
pip install --target ./package requests_aws4auth

すべての Lambda 実行環境に Boto3 がインストールされているため、デプロイパッケージに含める必要はありません。

アプリケーションコードや相互依存性をパッケージ化します。

cd package
zip -r ../lambda.zip .

cd ..
zip -g lambda.zip sample.py

[トリガーの追加] を選択し、[S3] を選択します。

バケットを選択します。

[イベントタイプ] で、[PUT] を選択します。

[プレフィックス] に logs/ と入力します。

[サフィックス] では、.log と入力します。

再帰呼び出しの警告を確認し、[追加] を選択します。

[～からアップロード] および [.zip ファイルをアップロード] を選択してから、デプロイパッケージをアップロードするプロンプトに従います。

アップロードが終了したら、[ランタイム設定] を編集し、[ハンドラ] を sample.handler に変更します。この設定により、トリガー後に実行するファイル (sample.py) およびメソッド (handler) が Lambda に通知されます。