はじめに

ローカル環境で動作しているPythonスクリプトを、AWS Lambdaでサーバーレスに自動実行できるように変換する方法を自分なりにまとめてみました。

本記事では、以前投稿した記事で扱ったMySQLからS3へのデータエクスポート処理を題材に、ローカル実行用のPythonスクリプトをLambda関数に変換する具体的な手順を学んでいきます。

元記事との違いについて
元記事ではローカル環境のMySQLからデータを取得していましたが、本記事ではLambda化を主題とするため、RDS for MySQLを使用します。RDSを使用することで、よりAWSネイティブな構成となり、実際の本番環境に近い形での学習が可能になります。

この記事で学べること：

RDS for MySQLの基本的なセットアップ
ローカル実行とLambda実行の違い
Lambda用コード変換の5つのポイント
Lambdaレイヤーによる外部ライブラリの管理
Lambda関数の作成とデプロイ手順
VPC・セキュリティグループの設定
環境変数とIAMロールの設定
テスト実行と動作確認

次の記事について：
Lambda関数の定期実行や監視設定については、次の記事「AWS EventBridgeでLambda関数を定期実行する方法」で詳しく実践していきます。

1. AWS Lambdaとは

AWS Lambdaは、サーバーを管理することなくコードを実行できるサーバーレスコンピューティングサービスです。

特徴	説明
サーバーレス	インフラの構築・管理が不要
イベント駆動	特定のイベントで自動実行
従量課金	実行時間とメモリ使用量に応じた課金
自動スケーリング	負荷に応じて自動的にスケール

1.2 Lambda化のメリット

運用面

サーバー管理不要で保守コスト削減
自動スケーリングで処理量の変動に対応
高可用性を自動的に確保

コスト面

実行時間分のみの従量課金
アイドル時のコスト発生なし
小規模処理では無料枠内で運用可能

開発面

デプロイが簡単
AWSサービスとの統合が容易
バージョン管理が標準で利用可能

2. 題材：RDS MySQLからS3へのデータエクスポート

2.1 処理の概要

本記事では、以下の処理をLambda化します。

1. RDS MySQLからデータを取得
2. Parquet形式に変換
3. S3バケットにアップロード

2.2 ローカル実行用のコード

まず、Lambda化する前のローカル実行用コードを確認します。

import boto3
import os
from dotenv import load_dotenv
import pandas as pd
import pymysql
from io import BytesIO
from datetime import datetime

def upload_mysql_to_s3(table_name):
    """MySQLテーブルをParquet形式でS3にアップロード"""
    try:
        # .envファイルから環境変数を読み込む
        load_dotenv()
        
        # S3クライアントを作成（認証情報を明示）
        s3_client = boto3.client(
            's3',
            aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
            aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
            region_name=os.getenv('AWS_REGION', 'ap-northeast-1')
        )
        
        # MySQL接続
        connection = pymysql.connect(
            host=os.getenv('MYSQL_HOST'),
            port=int(os.getenv('MYSQL_PORT', 3306)),
            user=os.getenv('MYSQL_USER'),
            password=os.getenv('MYSQL_PASSWORD'),
            database=os.getenv('MYSQL_DATABASE')
        )
        
        # データ取得
        query = f"SELECT * FROM {table_name}"
        df = pd.read_sql(query, connection)
        
        # Parquet形式に変換
        buffer = BytesIO()
        df.to_parquet(buffer, engine='pyarrow', compression='snappy', index=False)
        
        # S3にアップロード
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        s3_key = f'data/{table_name}/{timestamp}.parquet'
        
        s3_client.put_object(
            Bucket=os.getenv('S3_BUCKET_NAME'),
            Key=s3_key,
            Body=buffer.getvalue()
        )
        
        connection.close()
        print(f"✓ アップロード成功: {s3_key}")
        return True
        
    except Exception as e:
        print(f"✗ エラー: {e}")
        return False

# 実行
if __name__ == "__main__":
    upload_mysql_to_s3('users')

3. RDS for MySQLのセットアップ

Lambda関数からアクセスするRDSインスタンスを作成します。

3.1 RDSインスタンスの作成

1. データベースの作成

2. エンジンの選択

項目	設定値
エンジンタイプ	MySQL
エンジンバージョン	MySQL 8.0.x（最新）
テンプレート	開発/テスト

3. 設定

項目	設定値	説明
DBインスタンス識別子	`lambda-test-db`	インスタンス名
マスターユーザー名	`admin`	ルートユーザー
マスターパスワード	任意の強力なパスワード	8文字以上推奨

4. インスタンスの設定

項目	設定値	説明
DBインスタンスクラス	db.t3.micro	無料枠対象
ストレージタイプ	汎用SSD(gp3)	標準的な選択
ストレージ割り当て	20 GiB	最小値

5. 接続設定（重要）

項目	設定値	説明
VPC	デフォルトVPC	Lambda関数と同じVPC
パブリックアクセス	なし	セキュリティのため無効
VPCセキュリティグループ	新規作成	`lambda-rds-sg`
アベイラビリティーゾーン	優先なし	自動選択

6. データベース認証

「パスワード認証」を選択

7. 追加設定

項目	設定値
初期データベース名	`testdb`
バックアップ	自動バックアップを有効化
保持期間	7日

8. データベースの作成

「データベースの作成」ボタンをクリック
作成完了まで約5〜10分待機

3.2 RDS接続情報の確認

1. RDSコンソールで確認

RDS → データベース → lambda-test-db をクリック

2. エンドポイント情報を記録

項目	例
エンドポイント	`lambda-test-db.xxxxx.ap-northeast-1.rds.amazonaws.com`
ポート	`3306`

このエンドポイント情報は、後ほどLambda関数の環境変数に設定します。

3.3 サンプルデータの準備

RDSにテスト用のテーブルとデータを作成します。
今回はLambda化が主題のため、省略します。

過去記事で使用した以下のデータセットで作成しました。

4. ローカル実行とLambda実行の違い

4.1 主要な違い

Lambda化する際に意識すべきローカル環境とLambda環境を整理します。

項目	ローカル環境	Lambda環境
エントリーポイント	任意の関数名	`lambda_handler(event, context)` 必須
環境変数	`.env`ファイルから読み込み	Lambda環境変数から読み込み
AWS認証	アクセスキー明示指定	IAMロールで自動認証
外部ライブラリ	pip install	Lambdaレイヤーで提供
データベース接続	直接接続可能	VPC設定が必要
実行環境	ローカルマシン	AWS管理のコンテナ
戻り値	True/Falseなど自由	辞書形式（statusCode, body）
ログ出力	コンソール	CloudWatch Logs

4.2 Lambda化が必要な理由

ローカル環境の課題
    ↓
  - 実行環境の管理が必要
  - 定期実行には別途スケジューラが必要
  - スケールアップに手動対応が必要
  - 可用性の確保が困難

Lambda化による解決
    ↓
  - インフラ管理不要
  - EventBridgeで簡単に定期実行
  - 自動スケーリング
  - 高可用性を自動確保

5. Lambda用コード変換の実践

Lambda化するために必要な変更を、4つのポイント別で変更を加えます。

ポイント1: エントリーポイントの追加

変更点
Lambdaはlambda_handlerという関数名を起点として実行を開始するため、変更する。

変更前（ローカル実行用）

def upload_mysql_to_s3(table_name):
    # 処理...
    return True

# スクリプトの最後で直接呼び出し
if __name__ == "__main__":
    upload_mysql_to_s3('users')

変更後（Lambda実行用）

def lambda_handler(event, context):
    """
    Lambda関数のエントリーポイント
    
    Parameters:
        event (dict): Lambda呼び出し時のイベントデータ
        context (LambdaContext): Lambda実行環境の情報
    
    Returns:
        dict: HTTPステータスコードとレスポンスボディ
    """
    # 環境変数からテーブル名を取得
    table_name = os.getenv('TABLE_NAME', 'users')
    
    try:
        result = upload_mysql_to_s3(table_name)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success',
                'table': table_name,
                'success': result
            })
        }
    except Exception as e:
        print(f"エラー: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Error',
                'error': str(e)
            })
        }

def upload_mysql_to_s3(table_name):
    # 既存の処理をそのまま使用
    # ...

パラメータの説明

パラメータ	型	説明	例
`event`	dict	呼び出し元から渡されるデータ	`{}`
`context`	LambdaContext	Lambda実行環境の情報	関数名、残り実行時間など

eventパラメータを使った動的なテーブル名の指定方法は、次の記事「AWS EventBridgeでLambda関数を定期実行する方法」で詳しく解説します。

戻り値の形式

{
    'statusCode': 200,  # HTTPステータスコード
    'body': json.dumps({  # JSON文字列化が必要
        'message': 'Success'
    })
}

ポイント2: 環境変数読み込みの変更

変更点
環境変数から直接取得するように変更する。
(.envの読み込みの削除)

理由
① Lambda環境では.envファイルが存在しない
② AWS環境変数から直接読み込む方が安全で管理しやすい
③ KMSによる暗号化が可能

AWS環境変数は後ほど設定します

変更前（ローカル実行用）

from dotenv import load_dotenv

def upload_mysql_to_s3(table_name):
    # .envファイルを読み込む
    load_dotenv()
    
    # 環境変数を取得
    host = os.getenv('MYSQL_HOST')
    port = int(os.getenv('MYSQL_PORT', 3306))
    user = os.getenv('MYSQL_USER')
    # ...

変更後（Lambda実行用）

# dotenvのimportは削除

def upload_mysql_to_s3(table_name):
    # load_dotenv()の呼び出しは不要
    
    # 環境変数を直接取得
    host = os.getenv('MYSQL_HOST')
    port = int(os.getenv('MYSQL_PORT', 3306))
    user = os.getenv('MYSQL_USER')
    # ...

ポイント3: AWS認証方法の変更

変更点
IAMロールが付与され、boto3が認証情報を取得するため、認証情報の取得を明示的に行うコードの削除を行う

変更前（ローカル実行用）

# 認証情報を明示的に指定
s3_client = boto3.client(
    's3',
    aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
    aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
    region_name=os.getenv('AWS_REGION', 'ap-northeast-1')
)

変更後（Lambda実行用）

# IAMロールで自動認証（認証情報の指定不要）
s3_client = boto3.client('s3')

# またはリージョンのみ明示的に指定
s3_client = boto3.client('s3', region_name='ap-northeast-1')

認証フロー

Lambda関数起動
    ↓
IAMロール自動付与
    ↓
boto3が認証情報を自動取得
    ↓
AWSサービスへアクセス

ポイント4: 戻り値の形式変更

変更点
戻り値の形式をHTTPレスポンス形式に変更する。

理由
① Lambda関数は標準的なHTTPレスポンス形式で結果を返す必要がある
② API GatewayやEventBridgeとの統合を考慮
③ ログやモニタリングで成功・失敗を判別しやすくなる

変更前（ローカル実行用）

def upload_mysql_to_s3(table_name):
    try:
        # 処理...
        return True
    except Exception as e:
        print(f"エラー: {e}")
        return False

変更後（Lambda実行用）

def lambda_handler(event, context):
    try:
        result = upload_mysql_to_s3(table_name)
        
        # 成功時の戻り値
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success',
                'table': table_name,
                'timestamp': datetime.now().isoformat()
            })
        }
    except Exception as e:
        # エラー時の戻り値
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Error',
                'error': str(e)
            })
        }

HTTPステータスコード一覧

コード	意味	使用例
200	成功	正常に処理完了
400	クライアントエラー	パラメータ不正
403	権限エラー	IAMロール不足
500	サーバーエラー	予期しないエラー
503	サービス利用不可	外部サービス接続失敗

今回はLambda化がメインとなるため、エラーの種類に応じたレスポンスを返す実装は行っておりません。

変換後コード

4つのポイントをすべて適用したLambda用コードです。

import boto3
import os
import pandas as pd
import pymysql
from io import BytesIO
from datetime import datetime
import json

def lambda_handler(event, context):
    """
    Lambda関数のエントリーポイント
    RDS MySQLからS3へのデータエクスポート処理を実行
    """
    # 環境変数からテーブル名を取得
    table_name = os.getenv('TABLE_NAME', 'users')
    
    try:
        result = upload_mysql_to_s3(table_name)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success',
                'table': table_name,
                'success': result
            })
        }
    except Exception as e:
        print(f"エラー: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Error',
                'error': str(e)
            })
        }


def upload_mysql_to_s3(table_name):
    """
    RDS MySQLテーブルをParquet形式でS3にアップロード
    
    Args:
        table_name (str): 転送するテーブル名
    
    Returns:
        bool: 成功時True、失敗時例外をスロー
    """
    try:
        # S3クライアント作成（IAMロールで自動認証）
        s3_client = boto3.client('s3')
        
        # RDS MySQL接続（環境変数から直接取得）
        connection = pymysql.connect(
            host=os.getenv('MYSQL_HOST'),
            port=int(os.getenv('MYSQL_PORT', 3306)),
            user=os.getenv('MYSQL_USER'),
            password=os.getenv('MYSQL_PASSWORD'),
            database=os.getenv('MYSQL_DATABASE')
        )
        
        # データ取得
        query = f"SELECT * FROM {table_name}"
        df = pd.read_sql(query, connection)
        print(f"✓ データ取得完了: {len(df)}行")
        
        # Parquet形式に変換
        buffer = BytesIO()
        df.to_parquet(buffer, engine='pyarrow', compression='snappy', index=False)
        buffer.seek(0)
        
        # S3にアップロード
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        s3_key = f"data/{table_name}/{timestamp}.parquet"
        
        s3_client.put_object(
            Bucket=os.getenv('S3_BUCKET_NAME'),
            Key=s3_key,
            Body=buffer.getvalue(),
            ContentType='application/x-parquet'
        )
        
        print(f"✓ アップロード成功: s3://{os.getenv('S3_BUCKET_NAME')}/{s3_key}")
        print(f"  ファイルサイズ: {len(buffer.getvalue())} バイト")
        
        connection.close()
        return True
        
    except Exception as e:
        print(f"✗ エラー発生: {e}")
        raise e

6. Lambdaレイヤーによる外部ライブラリ管理

6.1 Lambdaレイヤーとは

Lambdaレイヤーは、Lambda関数で使用する外部ライブラリを別パッケージとして管理する仕組みです。

メリット

メリット	説明
デプロイサイズの削減	関数コード本体のサイズを小さく保てる
再利用性	複数のLambda関数で同じレイヤーを共有可能
更新の効率化	ライブラリのみの更新が容易
バージョン管理	ライブラリのバージョンを管理しやすい

6.2 必要なライブラリ

今回の処理で必要な外部ライブラリ：

ライブラリ	用途	提供方法
pandas	データフレーム操作	AWS公式レイヤー
pyarrow	Parquet形式への変換	AWS公式レイヤー
pymysql	MySQL接続	カスタムレイヤー（CloudShellで作成）

boto3はLambda環境に標準で含まれているため、レイヤーに含める必要はありません。

pandasとpyarrowはサイズが大容量ライブラリになるため、公式レイヤーを使用します。

6.3 AWS公式レイヤーの追加（pandas・pyarrow）

AWS SDK for pandas（旧AWS Data Wrangler）は、AWSが公式に提供しているレイヤーで、pandas、pyarrow、NumPyなどのデータ処理ライブラリが含まれています。

ステップ1: AWS公式レイヤーのARNを確認

各リージョンごとに専用のARNが用意されています。以下のリンクから最新のARNを確認してください。

公式ドキュメント：
https://aws-sdk-pandas.readthedocs.io/en/stable/layers.html

東京リージョン（ap-northeast-1）の例：

arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:15

バージョン番号（末尾の:15など）は定期的に更新されます。必ず公式ドキュメントで最新バージョンを確認してください。

ステップ2: Lambda関数にレイヤーを追加

1. Lambda関数ページを開く

AWSマネジメントコンソール → Lambda → 対象の関数

2. レイヤーセクションへ移動

関数ページを下にスクロール → 「レイヤー」セクション

3. レイヤーの追加

「レイヤーの追加」ボタンをクリック

4. ARNを指定してレイヤーを追加

項目	設定値
レイヤーソース	ARNを指定
ARN	上記で確認したARNを入力

例：

arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:15

5. 追加

「追加」ボタンをクリック

6.4 カスタムレイヤーの作成（pymysql）

pymysqlは軽量なライブラリなので、CloudShellを使って簡単にレイヤーを作成できます。

ステップ1: CloudShellでレイヤーを作成

1. CloudShellを起動

2. レイヤー用ディレクトリを作成

# 作業ディレクトリを作成
mkdir -p pymysql-layer/python

# pymysqlをインストール
pip install pymysql -t pymysql-layer/python/

# ZIPファイルに圧縮
cd pymysql-layer
zip -r ../pymysql-layer.zip python/
cd ..

# ファイルサイズを確認
ls -lh pymysql-layer.zip

実行結果の例：

Successfully installed pymysql-1.1.0
  adding: python/ (stored 0%)
  adding: python/pymysql/ (stored 0%)
  ...
-rw-r--r-- 1 cloudshell-user cloudshell-user 156K Jan 27 12:34 pymysql-layer.zip

ディレクトリ構造：

pymysql-layer/
└── python/          ← この名前は必須（Lambdaの命名規則）
    └── pymysql/

ステップ2: Lambda レイヤーを作成

CloudShellから直接作成

aws lambda publish-layer-version \
    --layer-name pymysql-layer \
    --description "PyMySQL library for Lambda" \
    --zip-file fileb://pymysql-layer.zip \
    --compatible-runtimes python3.11 \
    --compatible-architectures x86_64

実行結果の例：

{
    "LayerArn": "arn:aws:lambda:ap-northeast-1:123456789012:layer:pymysql-layer",
    "LayerVersionArn": "arn:aws:lambda:ap-northeast-1:123456789012:layer:pymysql-layer:1",
    "Version": 1,
    ...
}

ステップ3: Lambda関数にpymysqlレイヤーを追加

項目	設定値
レイヤーソース	カスタムレイヤー
カスタムレイヤー	`pymysql-layer`
バージョン	1（最新）

7. Lambda関数の作成とデプロイ

7.1 Lambda関数の作成

AWSコンソールでの作成手順

1. 関数の作成

Lambda コンソール → 「関数の作成」

2. 基本情報の入力

項目	設定値
作成方法	一から作成
関数名	`rds-mysql-to-s3-transfer`
ランタイム	Python 3.11
アーキテクチャ	x86_64

3. アクセス権限

「デフォルトの実行ロールを変更」を展開
「基本的なLambdaアクセス権限で新しいロールを作成」を選択

4. 詳細設定（重要）

「詳細設定」を展開
「VPC を有効にする」にチェック

VPC設定

項目	設定値
VPC	RDSインスタンスと同じVPC
サブネット	RDSと同じサブネット（複数選択推奨）
セキュリティグループ	後ほど設定

Lambda関数がRDSにアクセスするためには、同じVPC内に配置する必要があります。

5. 関数の作成

「関数の作成」ボタンをクリック

7.2 コードのデプロイ

1. コードエディタにコードを貼り付け

変換後のLambda用コードを貼り付け

2. Deployボタンをクリック

右上の「Deploy」ボタンをクリック

3. デプロイ完了を確認

Successfully deployed changes to function rds-mysql-to-s3-transfer

7.3 レイヤーの追加

1. AWS SDK for pandas レイヤーの追加

レイヤーセクション → 「レイヤーの追加」
レイヤーソース: ARNを指定
ARN: arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:15

2. pymysql レイヤーの追加

レイヤーセクション → 「レイヤーの追加」
レイヤーソース: カスタムレイヤー
カスタムレイヤー: pymysql-layer
バージョン: 1

8. VPCとセキュリティグループの設定

8.1 Lambda用セキュリティグループの作成

Lambda関数からRDSにアクセスするための通信を許可します。

1. EC2コンソールを開く

EC2 → セキュリティグループ → 「セキュリティグループを作成」

2. 基本情報

項目	設定値
セキュリティグループ名	`lambda-sg`
説明	Lambda function security group
VPC	RDSと同じVPC

3. アウトバウンドルール（デフォルトのまま）

タイプ	プロトコル	ポート範囲	送信先
すべてのトラフィック	すべて	すべて	0.0.0.0/0

4. セキュリティグループを作成

8.2 RDSセキュリティグループの更新

RDSがLambda関数からの接続を受け入れるように設定します。

1. RDS用セキュリティグループを開く

EC2 → セキュリティグループ → lambda-rds-sg（RDS作成時に作成したSG）

2. インバウンドルールを編集

「インバウンドルールを編集」をクリック

3. ルールの追加

タイプ	プロトコル	ポート範囲	ソース	説明
MySQL/Aurora	TCP	3306	`lambda-sg`	Lambda function access

4. ルールを保存

8.3 Lambda関数のVPC設定

1. Lambda関数の設定を開く

Lambda → rds-mysql-to-s3-transfer → 設定 → VPC

2. 編集

「編集」ボタンをクリック

3. VPC設定

項目	設定値
VPC	RDSと同じVPC
サブネット	プライベートサブネット（複数選択）
セキュリティグループ	`lambda-sg`

4. 保存

「保存」ボタンをクリック

VPC設定の反映には数分かかる場合があります。ステータスが「更新中」から「アクティブ」になるまで待ちましょう。

9. 環境変数の設定

9.1 環境変数とは

Lambda関数で使用する設定値（データベース接続情報など）を管理する仕組みです。

メリット

コードに機密情報を含めない
コード変更なしで設定変更可能
環境別（開発/本番）の設定を簡単に切り替え

9.2 必要な環境変数

キー	説明	例
`MYSQL_HOST`	RDSエンドポイント	`lambda-test-db.xxxxx.ap-northeast-1.rds.amazonaws.com`
`MYSQL_PORT`	MySQLポート番号	`3306`
`MYSQL_USER`	MySQL接続ユーザー名	`admin`
`MYSQL_PASSWORD`	MySQL接続パスワード	`your-password`
`MYSQL_DATABASE`	MySQLデータベース名	`testdb`
`S3_BUCKET_NAME`	転送先S3バケット名	`my-data-bucket`
`TABLE_NAME`	転送するテーブル名	`users`

9.3 設定手順

1. 設定タブへ移動

Lambda関数ページ → 「設定」タブ → 「環境変数」

2. 編集

「編集」ボタンをクリック

3. 環境変数の追加

「環境変数の追加」ボタンをクリック
上記のキーと値を順番に入力

環境変数の入力例：

MYSQL_HOST = lambda-test-db.c1234567890.ap-northeast-1.rds.amazonaws.com
MYSQL_PORT = 3306
MYSQL_USER = admin
MYSQL_PASSWORD = YourStrongPassword123!
MYSQL_DATABASE = testdb
S3_BUCKET_NAME = my-data-bucket
TABLE_NAME = users

4. 保存

「保存」ボタンをクリック

9.4 機密情報の暗号化（推奨）

パスワードなどの機密情報は暗号化することを強く推奨します。

1. 暗号化ヘルパーを有効化

環境変数編集画面で「暗号化設定」を展開

2. KMSキーの選択

「転送時の暗号化用のヘルパーの有効化」にチェック
デフォルトのキーaws/lambdaを選択

3. 暗号化する環境変数を選択

MYSQL_PASSWORDの横にある「暗号化」ボタンをクリック

10. IAMロールの設定

10.1 IAMロールとは

Lambda関数がAWSサービス（S3、VPC、RDSなど）にアクセスするための権限セットです。

Lambda関数 → IAMロール（実行ロール） → AWSサービスへのアクセス権限

10.2 必要な権限

サービス	権限	用途
S3	`s3:PutObject`	Parquetファイルのアップロード
CloudWatch Logs	`logs:CreateLogGroup`	ログ記録
VPC	`ec2:CreateNetworkInterface`	VPC内のRDSアクセス
VPC	`ec2:DescribeNetworkInterfaces`	ネットワークインターフェース情報取得
VPC	`ec2:DeleteNetworkInterface`	ネットワークインターフェース削除

10.3 IAMロールの設定手順

S3書き込み権限の追加

1. IAMロールを開く

Lambda関数ページ → 「設定」→ 「アクセス権限」
「ロール名」のリンクをクリック

2. ポリシーのアタッチ

「ポリシーをアタッチ」ボタンをクリック

3. S3ポリシーの選択

検索ボックスにS3と入力
AmazonS3FullAccessにチェック
（本番環境では最小権限の原則に従い、カスタムポリシーを推奨）

4. VPCポリシーの確認

Lambda関数作成時に自動的に AWSLambdaVPCAccessExecutionRole が付与されているか確認

5. アタッチ

「ポリシーをアタッチ」ボタンをクリック

10.4 カスタムポリシーの作成（推奨）

セキュリティのため、最小限の権限のみを付与するカスタムポリシーを作成します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:PutObjectAcl"
      ],
      "Resource": "arn:aws:s3:::your-data-bucket/data/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "ec2:CreateNetworkInterface",
        "ec2:DescribeNetworkInterfaces",
        "ec2:DeleteNetworkInterface"
      ],
      "Resource": "*"
    }
  ]
}

カスタムポリシーの適用手順

IAMコンソール → 「ポリシー」→「ポリシーを作成」
「JSON」タブをクリック
上記のJSONをペースト（バケット名を実際の値に変更）
ポリシー名を入力（例: LambdaRDSToS3Policy）
「ポリシーを作成」をクリック
Lambda関数のIAMロールに戻ってアタッチ

11. タイムアウトとメモリの調整

11.1 デフォルト設定の制限

Lambda関数のデフォルト設定では、データ処理に不十分な場合があります。

設定項目	デフォルト値	推奨値（本題材の場合）
タイムアウト	3秒	5分（300秒）
メモリ	128 MB	512 MB～1024 MB

11.2 設定変更手順

1. 一般設定を開く

Lambda関数ページ → 「設定」→「一般設定」→「編集」

2. 設定値の変更

項目	設定値	理由
メモリ	512 MB	pandasとpyarrowがメモリを消費
タイムアウト	5分（300秒）	RDS接続とデータ変換に時間がかかる可能性
一時ストレージ	512 MB（デフォルト）	Parquetファイル変換用

3. 保存

「保存」ボタンをクリック

11.3 メモリとCPUの関係

重要な仕組み：Lambdaではメモリ割り当てに比例してCPU性能も向上します。

メモリ: 128 MB  → CPU: 低
メモリ: 512 MB  → CPU: 中
メモリ: 1024 MB → CPU: 高
メモリ: 1792 MB → CPU: vCPU 1個相当

最適化のポイント

処理時間が長い → メモリを増やす（CPU性能向上）
メモリエラーが発生 → メモリを増やす
コスト最適化 → 必要最小限のメモリに調整

12. テスト実行と動作確認

12.1 手動テスト実行

Lambda関数が正しく動作するか、手動でテストを実行します。

1. テストタブを開く

Lambda関数ページ → 「テスト」タブ

2. テストイベントの作成

「新しいイベントを作成」を選択

3. イベント情報の入力

項目	設定値
テンプレート	hello-world
イベント名	`test-event`

イベントJSON

{}

現時点では空のイベントでテストします。次の記事で、イベントにパラメータを含める方法を解説します。

4. 保存して実行

「保存」→「テスト」ボタンをクリック

12.2 実行結果の確認

成功時の出力例

{
  "statusCode": 200,
  "body": "{\"message\": \"Success\", \"table\": \"users\", \"success\": true}"
}

テーブル名の確認

body内のtableフィールドで、環境変数TABLE_NAMEに設定したテーブル名が処理されていることを確認できます

実行ログの確認

START RequestId: 12345678-1234-1234-1234-123456789012 Version: $LATEST
✓ データ取得完了: 5行
✓ アップロード成功: s3://my-bucket/data/users/20250127_123456.parquet
  ファイルサイズ: 1234 バイト
END RequestId: 12345678-1234-1234-1234-123456789012
REPORT RequestId: 12345678-1234-1234-1234-123456789012
Duration: 2345.67 ms
Billed Duration: 2346 ms
Memory Size: 512 MB
Max Memory Used: 234 MB

12.3 S3での結果確認

1. S3コンソールを開く

AWSマネジメントコンソール → S3 → 該当のバケット

2. ファイルの確認

data/{table_name}/ ディレクトリ配下にParquetファイルが存在することを確認

ファイルパス例

s3://my-data-bucket/data/users/20250127_123456.parquet

3. ファイルの内容確認（オプション）

Amazon Athenaを使用してParquetファイルの内容を確認できます。

-- Athenaでテーブルを作成
CREATE EXTERNAL TABLE users_parquet (
  id INT,
  name STRING,
  email STRING,
  created_at TIMESTAMP
)
STORED AS PARQUET
LOCATION 's3://my-data-bucket/data/users/';

-- データを確認
SELECT * FROM users_parquet LIMIT 10;

13. CloudWatch Logsでの基本的なログ確認

13.1 CloudWatch Logsとは

Lambda関数の実行ログを自動的に記録するAWSのログ管理サービスです。

記録される情報

print()で出力した内容
実行時間、メモリ使用量
発生したエラー

13.2 ログの確認手順

1. モニタリングタブを開く

Lambda関数ページ → 「モニタリング」タブ

2. CloudWatch Logsを表示

「CloudWatch のログを表示」ボタンをクリック

3. ロググループの確認

/aws/lambda/rds-mysql-to-s3-transfer というロググループが表示される

4. ログストリームの選択

最新のログストリーム（日時が最新のもの）をクリック

13.3 ログの見方

正常実行時のログ例

2025-01-27T12:34:56.789Z  START RequestId: abc123...
2025-01-27T12:34:57.123Z  ✓ データ取得完了: 5行
2025-01-27T12:34:58.456Z  ✓ アップロード成功: s3://bucket/data/users/20250127_123456.parquet
2025-01-27T12:34:58.457Z    ファイルサイズ: 1234 バイト
2025-01-27T12:34:58.789Z  END RequestId: abc123...
2025-01-27T12:34:58.789Z  REPORT RequestId: abc123...
    Duration: 2000.00 ms
    Billed Duration: 2000 ms
    Memory Size: 512 MB
    Max Memory Used: 234 MB

エラー発生時のログ例

2025-01-27T12:34:56.789Z  START RequestId: abc123...
2025-01-27T12:34:57.123Z  エラー: (2003, "Can't connect to MySQL server")
2025-01-27T12:34:57.125Z  END RequestId: abc123...

CloudWatch Logsを使った詳細な監視とアラート設定については、次の記事「AWS EventBridgeでLambda関数を定期実行する方法」で解説します。

14. まとめ

本記事では、RDS MySQLからS3へのデータエクスポート処理を題材に、ローカル実行用のPythonスクリプトをAWS Lambda化する方法を学びました。

重要なポイント

1. RDSの利用

ローカルMySQLの代わりにRDS for MySQLを使用
Lambda関数との統合がAWSネイティブで簡単
VPC・セキュリティグループによるセキュアな接続

2. コード変換の5つのポイント

lambda_handlerのエントリーポイント追加
.envファイルからAWS環境変数への移行
IAMロールによる自動認証への変更
辞書形式の戻り値への統一

3. インフラ設定

Lambdaレイヤーでの外部ライブラリ管理
VPC内でのLambdaとRDSの接続設定
セキュリティグループによるアクセス制御
IAMロールによる最小権限の原則
タイムアウトとメモリの適切な設定

4. 動作確認

手動テスト実行での動作確認
CloudWatch Logsでのログ確認
S3へのファイル出力確認