はじめに
この記事では、AWS Lambdaを使用して、特定のS3イベントに応じてGlueジョブをトリガーし、その結果をSNSで通知する方法を紹介します。コード例には、すべてダミーの情報が含まれています。
はじめに
AWS Lambdaは、イベントドリブンなサーバーレスコンピューティングサービスです。これにより、インフラストラクチャの管理を行わずにコードを実行できます。AWS Glueは、ETL（Extract, Transform, Load）ジョブを実行するためのマネージドサービスであり、AWS SNSは通知サービスです。
以下のコードでは、S3にファイルがアップロードされると、LambdaがトリガーされてGlueジョブを開始し、結果をSNSで通知します。
コードの詳細
必要なライブラリのインポート
import boto3
from urllib.parse import unquote
import logging
import time
-
boto3: AWS SDK for Python。AWSサービスと対話するために使用します。 -
unquote: URLエンコードされた文字列をデコードするために使用します。 -
logging: ログの設定。 -
time: ジョブのステータスをポーリングするために使用します。
ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)
- ログのレベルをINFOに設定します。
Lambdaハンドラー関数
def lambda_handler(event, context):
glue_client = boto3.client('glue')
s3_client = boto3.client('s3')
sns_client = boto3.client('sns')
-
glue_client,s3_client,sns_clientを作成して、それぞれのAWSサービスと対話します。
SNSトピックのマッピング
sns_topics = {
'folder1': 'arn:aws:sns:ap-northeast-1:123456789012:folder1_topic',
'folder2': 'arn:aws:sns:ap-northeast-1:123456789012:folder2_topic',
'folder3': 'arn:aws:sns:ap-northeast-1:123456789012:folder3_topic'
}
- 各フォルダに対応するSNSトピックのARNをマッピングします。
イベントの処理
try:
logger.info("Event: %s", event)
if 'Records' not in event:
raise KeyError("Records key not found in event")
records = event['Records']
if not records or len(records) == 0:
raise ValueError("No Records found in event")
s3_event = records[0].get('s3')
if not s3_event:
raise ValueError("No S3 data found in event")
bucket_name = s3_event['bucket']['name']
encoded_key = s3_event['object'].get('key')
if not encoded_key:
raise ValueError("No object key found in S3 event")
key = unquote(encoded_key)
logger.info("Processed key: %s", key)
- イベントから必要な情報を抽出し、S3のバケット名とオブジェクトキーを取得します。
Glueジョブの起動
folder_name = key.split('/')[0]
file_name = key.split('/')[-1]
if key == 'folder1/file1.csv':
job_name = 'job1'
elif key == 'folder1/file2.csv':
job_name = 'job2'
elif key == 'folder2/file3.csv':
job_name = 'job3'
elif key == 'folder2/file4.csv':
job_name = 'job4'
elif key == 'folder3/file5.csv':
job_name = 'job5'
else:
logger.warning("No matching job for the key: %s", key)
return {'message': 'フォルダ名が間違っています'}
response = glue_client.start_job_run(JobName=job_name)
job_run_id = response['JobRunId']
logger.info("Job started successfully: %s", job_run_id)
- キーに基づいてGlueジョブを選択し、ジョブを開始します。
ジョブのステータスを確認
while True:
job_status = glue_client.get_job_run(JobName=job_name, RunId=job_run_id)
status = job_status['JobRun']['JobRunState']
logger.info("Current job status: %s", status)
if status in ['SUCCEEDED', 'FAILED', 'STOPPED']:
break
time.sleep(20)
- ジョブのステータスをポーリングし、完了するまで待ちます。
SNS通知の送信
sns_topic_arn = sns_topics[folder_name]
if status == 'SUCCEEDED':
message = f"{job_name.split('_')[0]}ジョブが成功しました。"
subject = f"{job_name.split('_')[0]}ジョブ成功"
else:
message = f"{job_name.split('_')[0]}ジョブが失敗しました。ステータス: {status}."
subject = f"{job_name.split('_')[0]}ジョブ失敗"
if sns_topic_arn:
sns_client.publish(
TopicArn=sns_topic_arn,
Subject=subject,
Message=message
)
- ジョブの結果に応じてSNS通知を送信します。
処理完了後のファイル削除
s3_client.delete_object(Bucket=bucket_name, Key=key)
logger.info("Deleted file: %s", key)
return {'message': '処理が完了しました'}
except Exception as e:
logger.error("Error processing event: %s", str(e))
raise e
- 処理が完了したファイルをS3から削除します。
結論
このLambda関数は、S3にファイルがアップロードされるとGlueジョブを起動し、その結果をSNSで通知する仕組みを提供します。これにより、データ処理の自動化と監視が容易になります。
このサンプルコードを参考に、自身のユースケースに合わせて適宜変更し、実装してみてください。