AWS Glue と Lambda のテクノロジー・アセスメントサーバレスETLのススメ

# AWS Glue Sample code
import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.transforms import SelectFields
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext

# 初期化
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# Jobインスタンス生成
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# S3からCSVを読み込み
input_path = "s3://your-bucket/input/"
df = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": [input_path]},
    format="csv",
    format_options={"withHeader": True}
)

# データ加工処理
processed_df = SelectFields.apply(df, paths=["id", "name", "created_at"])

# S3へParquetを出力
output_path = "s3://your-bucket/output/"
glueContext.write_dynamic_frame.from_options(
    frame=processed_df,
    connection_type="s3",
    connection_options={"paths": [output_path]},
    format="parquet"
)

job.commit()

# AWS Lambda Sample code
import boto3
import pandas as pd
import io

s3 = boto3.client('s3')
BUCKET = 'your-bucket'
INPUT_KEY = 'input/data.csv'
OUTPUT_KEY = 'output/result.parquet'

def lambda_handler(event, context):
    # S3からCSVを読み込み
    obj = s3.get_object(Bucket=BUCKET, Key=INPUT_KEY)
    df = pd.read_csv(io.BytesIO(obj['Body'].read()))
    
    # データ加工処理
    processed_df = df[['id', 'name', 'created_at']]
    
    # Parquet形式に変換
    out_buffer = io.BytesIO()
    processed_df.to_parquet(out_buffer, index=False, engine='pyarrow')

    # ParquetをS3へアップロード
    s3.put_object(
        Bucket=BUCKET,
        Key=OUTPUT_KEY,
        Body=out_buffer.getvalue(),
        ContentType='application/octet-stream'
    )
    
    return {
        'statusCode': 200,
        'body': 'OK'
    }