aws glue

date: 2024-04-26 excerpt: aws glueの概要と使い方

tag: aws aws glue

aws glueの概要と使い方

概要

ETL（Extract, Transform, Load）サービス
S3のデータをRedshiftやRDSなどにETLする
pythonのスクリプトを使ってETL処理を行う

使い方

S3にデータをアップロード
クローラーを作成してデータのスキーマを取得
pythonのスクリプトを作成しS3へアップロード
ジョブを作成してETL処理を行う
ジョブの実行結果を確認

1. S3にデータをアップロード

2. クローラーを作成してデータのスキーマを取得

IAMロールを作成
- AWSGlueServiceRole, S3FullAccess のポリシーをアタッチ

$ aws glue create-crawler \
    --name "example-crawler" \
    --role "arn:aws:iam::123456789012:role/YourGlueServiceRole" \
    --database-name "example-database" \
    --targets '{"S3Targets":[{"Path":"s3://example-bucket/data/"}]}'

3. pythonのスクリプトを作成しS3へアップロード

スクリプト例

import sys
import pandas as pd
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col
from awsglue.dynamicframe import DynamicFrame


args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://example-bucket/data/"]},
    format="csv",
    format_options={"withHeader": True},
    transformation_ctx="datasource0"
)

# DynamicFrameをPandas DataFrameに変換
df = datasource0.toDF().toPandas()

# Pandasで簡単なデータ操作（例：売上に10%の税を加える）
df['Sales'] = pd.to_numeric(df['Sales'], errors='coerce')
df['TaxedSales'] = df['Sales'] * 1.1

# Pandas DataFrameをSpark DataFrameに変換
spark_df = spark.createDataFrame(df)

# 変換したSpark DataFrameをDynamicFrameに変換して出力
dynamic_frame = DynamicFrame.fromDF(spark_df, glueContext, "dynamic_frame")

datasink4 = glueContext.write_dynamic_frame.from_options(
    frame=dynamic_frame,
    connection_type="s3",
    connection_options={"path": "s3://example-bucket/results/"},
    format="json",
    transformation_ctx="datasink4"
)

job.commit()

スクリプトのアップロード

$ aws s3 cp script.py s3://example-bucket/scripts/script.py

4. ジョブを作成してETL処理を行う

$ aws glue create-job \
    --name "example-job" \
    --role "arn:aws:iam::123456789012:role/YourGlueServiceRole" \
    --command '{"Name":"glueetl","ScriptLocation":"s3://example-bucket/scripts/script.py","PythonVersion":"3"}' \
    --default-arguments '{"--TempDir":"s3://example-bucket-temp/","--job-bookmark-option":"job-bookmark-enable"}'
$ aws glue start-job-run --job-name "example-job"

5. ジョブの実行結果を確認

$ aws glue list-jobs # ジョブの一覧を表示
$ aws glue get-job-run --job-name "example-job" --run-id "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" # ジョブの実行結果を表示