【BI】EMR+RedShiftでTableauインフラ環境を構築
今回は優秀なBIツールである「Tableau」のインフラ環境を、EMRとRedShiftで作ってみましょう。
なお、「BI」という単語に馴染みがないかもしれませんが、「BI」とは『Business Inteligence』の略となります。
つまりは仕事上の意思決定に必要な情報、それらをわかりやすいように分析・表示してくれるツールが「BIツール」です。
もしBIツールを使いこなすことができれば、重要な決定を正しく下したり、提案の有効性を強力にアピールすることが出来ます。
それでは早速構築方法を見ていきましょう。
概要
以下のような流れで構築を進めていきます。
Tableau分析をするためのインフラ基盤
EMRでデータ変換を行う
データ変換はEMR Python Libraryを使用する
Redshift上にデータをロードする
構成図
全体の構成図は以下の通りです。
開発する環境
まずはAWSで簡易的な分析基盤を開発していきます。
EMR(serverless)
RedShift
Lambda
アップロードされたCSVをロードできる形式に変換する
変換された形式のファイルをRedshiftにロードする
※ なお今回の環境構築はTerraformを使用して構築しております。
作成するスクリプト
後述のリポジトリを利用するために、以下の開発が必要となります。
アップロードされたCSVをロードできる形式にCSV形式にするPythonスクリプト
変換されたCSVをロードするLambda関数
どちらもリンクがサンプルになりますのでご利用ください。
解説
EMR
AWS EMR（Amazon Elastic MapReduce）は、Amazonが提供するビッグデータ解析サービスです。このサービスを使うことで、企業は大量のデータを迅速かつ効率的に分析することができます。データの解析を通じて、顧客の動向を把握したり、市場のトレンドを分析したりすることが可能になります。
利用料金は使った分だけ支払う形式なので、コストを抑えながら必要な時に必要なリソースを利用できる点が魅力です。また、他のAmazonのサービスと組み合わせることで、データの収集から解析までを一元管理することも可能です。このように、EMRはデータを活用してビジネスの意思決定を支援する強力なツールです。
RedShift
Amazon Redshiftは、AWSが提供するクラウドベースのデータウェアハウスサービスです。このサービスは、大量のデータを迅速かつ効率的に分析するためのもので、企業がより良い意思決定を行うための洞察を得ることを支援します。
またAmazon S3やDynamoDBなど、他のAWSサービスとスムーズに連携できます。
これらの特性により、Redshiftはマーケティングデータの分析、財務報告、顧客行動の分析など、多岐にわたるビジネスニーズに対応する強力なツールとなっています。
EMR Python Library
Amazon EMR ServerlessアプリケーションでPySparkジョブを実行するとき、依存関係として様々なPythonライブラリをパッケージ化することができます。しかし、標準のimageにはinstallされていませんので、Pythonのネイティブ機能を使用するか、仮想環境を構築する必要があります。
今回はDockerでImageを作成する手順を解説します。
構成ファイルは以下になります。
emr_python_library
├── Dockerfile
└── requirements.txtDockerfileの中身は以下になります。
FROM --platform=linux/x86_64 amazonlinux:2 AS base
RUN yum install -y python3
ENV VIRTUAL_ENV=/opt/venv
RUN python3 -m venv $VIRTUAL_ENV
ENV PATH="$VIRTUAL_ENV/bin:$PATH"
COPY requirements.txt .
RUN python3 -m pip install --upgrade pip && \
python3 -m pip install \
great_expectations==0.15.6 \
venv-pack==0.2.0
RUN python3 -m pip install -r requirements.txt
RUN mkdir /output && venv-pack -o /output/pyspark_ge.tar.gz
FROM scratch AS export
COPY --from=base /output/pyspark_ge.tar.gz /requirements.txtではpysparkに必要なライブラリを用意します。
numpy==1.21.6
pandas==1.3.5
pyarrow==12.0.1
scipy==1.7.3
matplotlib==3.5.3
pyspark==3.4.3以下の手順でpyspark_ge.tar.gzを作成します。
cd emr_python_library
docker build --output . .これによりファイル``pyspark_ge.tar.gz``が作成されます。
emr_python_library
├── Dockerfile
├── pyspark_ge.tar.gz
└── requirements.txt作成した``pyspark_ge.tar.gz``はS3にアップロードされるようにします。
resource "aws_s3_object" "script_venv_object" {
bucket = aws_s3_bucket.script_bucket.bucket
key = local.pyspark_env_archive # アップロードするファイルのS3上のパス
source = "./emr_python_library/${local.pyspark_env_archive}" # ローカルシステム上のファイルのパス
# オプション: ファイルの内容が変わったときだけアップロードをトリガーする
etag = filemd5("./emr_python_library/${local.pyspark_env_archive}")
}EMRでのデータ変換
今回はS3にPUTされたCSVファイルEventTriggerが検知して、EMR上で変換スクリプトを実行します。変換後別のS3にPUTします。
以下はサンプルのスクリプトです。
import sys
import urllib
from pyspark.sql import SparkSession
from pyspark.sql.functions import \
col, \
split, \
concat_ws, \
concat, \
date_format, \
lit
def main():
input_bucket = sys.argv[1]
input_key = urllib.parse.unquote_plus(sys.argv[2])
input_path = f"s3://{input_bucket}/{input_key}"
output_bucket = sys.argv[3]
output_key = urllib.parse.unquote_plus(sys.argv[4])
output_path = f"s3://{output_bucket}/{output_key}"
print(f"Input path: {input_path}")
print(f"Output path: {output_path}")
spark = SparkSession.builder.appName("S3 to EMR Serverless Processing").getOrCreate()
# S3からCSVファイルを読み込む
df = spark.read.option("encoding", "Shift_JIS").csv(input_path, header=True, inferSchema=True)
# split_consultday = split(col("consultday"), " ")
# データ処理（例：列の追加）
# df_modified = df....
# 列名の変更
# df_modified = df_modified...
# データ処理（例：特定の列を選択）
processed_df = df_modified...
# 処理結果をS3に保存
processed_df.write.option("encoding", "UTF-8").csv(output_path, header=True, mode="overwrite")
spark.stop()
if __name__ == "__main__":
main()Load Redshiftについて
変換されたCSVファイルがPUTされたら、別のEventTriggerでロード用のLambdaを実行します。
import boto3
import os
import urllib
def lambda_handler(event, context):
print(event)
print(context)
redshift = boto3.client('redshift-data')
cluster_identifier = os.environ.get('REDSHIFT_CLUSTER_IDENTIFIER')
database = os.environ.get('DATABASE_NAME')
db_user = os.environ.get('DATABASE_USER')
# イベントからファイル名とバケット名を取得
bucket = event['Records'][0]['s3']['bucket']['name']
key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'])
s3_path = f's3://{bucket}/{key}'
table_name = os.environ.get('TABLE_NAME')
redshift_iam_role_arn = os.environ.get('REDSHIFT_IAM_ROLE_ARN')
copy_query = f"""
COPY {database}.public.\"{table_name}\"
FROM '{s3_path}'
IAM_ROLE '{redshift_iam_role_arn}'
FORMAT AS CSV DELIMITER ',' QUOTE '"'
IGNOREHEADER 1
BLANKSASNULL
DATEFORMAT 'auto'
REGION AS 'ap-northeast-1'"""
print(f'copy_query: {copy_query}')
# Redshiftにクエリを送信
redshift.execute_statement(
ClusterIdentifier=cluster_identifier,
Database=database,
DbUser=db_user,
Sql=copy_query
)これでTableauインフラ環境が構築できました！
その他参考資料など
本記事で紹介されたコードは、下記を参考に記述されています。
Using Python libraries with EMR Serverless
：https://docs.aws.amazon.com/ja_jp/emr/latest/EMR-Serverless-UserGuide/using-python-libraries.html
以下のgithubからもソースコードを見ることが出来ます。
ソースコード：https://github.com/TodoONada/redshift-emr-infra
また、弊社では『マッチングワン』という『低コスト・短期にマッチングサービスを構築できる』サービスを展開しており、今回ご紹介するコードは、その『マッチングワン』でも使われているコードとなります。
本記事で紹介したようなプロダクトを開発されたい場合は、是非お問い合わせください。
またTodoONada株式会社では、この記事で紹介した以外にも、Supabase・Next.jsを使ったアプリの作成方法等についてご紹介しています。
下記記事にて一覧を見ることができますので、ぜひこちらもご覧ください！
お問合せ＆各種リンク
お問合せ：GoogleForm
ホームページ：https://libproc.com
運営会社：TodoONada株式会社
presented by
サポートしていただくと、筆者のやる気がガンガンアップします！