【AWS】redshift,S3をglueのETLでつなぐ

はじめに

KaggleやNishikaといったサービスでは整った形のデータセットがダウンロード可能でPythonですぐに読み込む事ができますが、本来こういった大規模なデータはどういった構造で蓄積されると良いのか興味がありました。

※過去にNishikaでデータ分析に入門してみたQiita記事

今回は色々な方の記事を参考にAWSのサービスを学びながら実践的な構成を目指したいと思います。
初心者ですので間違ってたらすいません。

データ分析基盤の構成

サイト情報を元にデータ分析に必要な基盤について教えて頂きました。

下記YOUTUBE動画で概要が分かりやすく説明されており、対応するAWSサービスも紹介されていました。

またこちらのサイトではデータレイク、データウェアハウス、データマートの各種特徴について分かりやすくまとめて頂いています。

lakeel.com：クラウド型アプリケーション開発・運用基盤より

今回の内容

今後サイトを参考に難易度を段階的に上げながらAWSの各種サービスを試していこうかと思います。
今回は各種AWSサービスの基本的な連携方法について以下を作成しながら勉強しようと思います。

1. S3に対するGlueのETL実行

2. S3に対するRedshiftからの接続

1．S3に対するGlueのETL

S3間のETLをGlueで作成し簡単なデータ処理を設定しようと思います。

• S3でデータレイク、データウェアハウス2つのバケットを作成しデータを設置
• GlueでETLに見立てたPython関数の定期実行を設定する
• データレイクのCSVから抽出したデータをデータウェアハウスにCSV形式で出力する

実施手順

データレイクに設置したCSV形式のタイタニック号の乗員、乗客データから、乗組員ID、生存情報(生存でTrue)、クラス、性別、年齢のカラムのみ抽出したデータをCSV形式でデータウェアハウスに出力する処理をGlue jobとして作成していきます。

S3にバケットを作成しデータを設置

Kaggleでダウンロードしたタイタニック号の乗員名簿データがCSV形式でDatalakeに置いてあるような状態を想定しています。

上記をデータレイク側のS3にタイタニック号データを設置します。

GlueにPythonスクリプトを設定

Gluejobを作成しスクリプトタブで以下の関数を設定します。

glue:script

import pandas
import s3fs
# s3などのAWSストレージに設置したライブラリファイルをjob detailタブで読み込むことも可能

source_s3_file = "s3://データレイクのバケット名/titanic/test.csv"
dest_s3_file ="s3://データウェアハウスのバケット名/test-etl.csv"

titanic_df = pandas.read_csv(source_s3_file)

# 乗組員ID、生存情報、クラス、性別、年齢のみ取り出す
titanic_df = titanic_df[['PassengerId', 'Survived', "Pclass", "Sex", "Age"]]

titanic_df.to_csv(dest_s3_file)

Runでスクリプトをテスト実行すると指定したS3バケットに特定のカラムのみ抽出されたCSVが出力されます。

定期実行を行う際はScheduleタブで実行頻度や時間を設定します。

実行結果はRunsタブのCloudwatch logsから実行したGlueスクリプトに対応するロググループが作成され閲覧できます。

※作成したGlue JobのIAM Roleには「S3へのアクセス」、「Cloud Watchへの書き込み」権限が必要でした。

データウェアハウス側にCSVが出力されればPythonによるETL処理の実行を確認できます。

2. S3に対するRedshiftからの接続

下記のハンズオン記事を参考に実施しました。

動画ではQuickSightを使いデータの集計結果の視覚化を行っていましたがそこまではやらなかったです。

• S3にデータレイクを作成しtxt型データを設置する
• Redshiftにデータ構造に対応したテーブルを作成する
• クエリ―の実行によりS3のデータをReadshiftに挿入する

実施手順

Redshiftにデータベース、各種テーブルを作成しS3に設置したレコード情報を含むテキストファイルからQuery Editorを使用しレコードを作成していきたいと思います。

S3にテキストデータを設置

先ほど利用したS3データレイクに各種Redshiftのレコードとなる元のテキストデータを設置しました。

テキストデータは以下のデータ構造に対応するレコードが含まれています。

手を動かしなら学ぶ Analytics サービス入門より

Redshiftでデータベース作成

ワークグループに関する各種設定はデフォルトのものを使用し、VPC、サブネット、セキュリティグループ等の設定も一旦デフォルトで作成しました。

ネットワーク関連の設定も自分なりにカスタマイズして勉強したかったのですが今回は行いませんでした。

作成したリソースは名前空間で構造化されるようです。

下記の記事ではRedshiftに関する構造や操作方法を参考にさせて頂きました。

クエリの実行

RedshiftメニューからRedshift query editor v2にアクセスし作成したDBに対しクエリの実行を行う事ができます。

ハンズオンで配布された資料をもとにテーブル構造を定義します。

CreateUser

create table users(
  userid integer not null,
  username char(8),
  firstname varchar(30),
  lastname varchar(30),
  city varchar(30),
  state char(2),
  email varchar(100),
  ...省略
  ) 

CreateDate

create table date(
  dateid smallint not null,
  caldate date not null,
  day character(3) not null,
  week smallint not null,
  month character(5) not null,
  qtr character(5) not null,
  year smallint not null,
  holiday boolean default('N'));

CreateSale

create table sales(
  salesid integer not null,
  listid integer not null,
  sellerid integer not null,
  buyerid integer not null,
  eventid integer not null,
  dateid smallint not null,
  qtysold smallint not null,
  pricepaid decimal(8,2),
  commission decimal(8,2),
  saletime timestamp);

Runで各クエリを実行するとテーブルが作成されます。

同様にS3に接続しデータ内容からレコードを作成するクエリが実行できます。

COPY句はRedshift固有のクエリでS3からCSVやtxt形式のデータをレコードとして読み込み可能との事です。

各テキストファイルのフィールドの区切りは"|"、もしくは"\t(タブ)"で、
時間のフォーマットは"MM/DD/YYYY HH:MI:SS"となっています。

user ETL

COPY users from 's3://S3データレイクの名前/data-analysis-handson/allusers_pipe.txt' iam_role 'IAMロールのARN' delimiter '|' ;

date ETL

COPY date from 's3://S3データレイクの名前/data-analysis-handson/date2008_pipe.txt' iam_role 'IAMロールのARN' delimiter '|' ;

sale ETL

COPY sales from 's3://S3データレイクの名前/data-analysis-handson/sales_tab.txt' iam_role 'IAMロールのARN' delimiter '\t'  timeformat 'MM/DD/YYYY HH:MI:SS';

※IAMロールにはあらかじめ作成されたRedshiftFullAccessを持つロールを設定しました。

Runでクエリを実行すればレコードが作成されます。

usersテーブルの内容を見るとレコードが作成されているのが確認できます。

まとめ

S3,Glue,Redshiftを使用したデータのやり取りを少しだけ実践してみました。
次は実戦的な構成を目指して勉強してみたいと思います。

S3に対するGlueのETL

• S3でデータレイク、データウェアハウス2つのバケットを作成しデータを設置
• GlueでETLに見立てたPython関数の定期実行を設定する
• データレイクのCSVから抽出したデータをデータウェアハウスにCSV形式で出力する

S3に対するRedshiftからの接続

• S3にデータレイクを作成しtxt型データを設置する
• Redshiftにデータ構造に対応したテーブルを作成する
• クエリ―の実行によりS3のデータをReadshiftに挿入する