AWS GlueのJob Bookmarkの使い方
実際にETLで処理するケースとしては、1日1回定期的に処理するなどのケースが多いと思います。
この場合、追加分のみを抽出してETL処理をする必要があります。
Glueには、前回どこまで処理したかを管理するJob Bookmarksという機能があります。
今回はこのJob Bookmarksを使ってみたいと思います。
準備
以下のようなフォーマットのapacheログをそれぞれ10行程度ずつ2ファイル用意します。
200.69.224.58 - - [16/Jan/2018:18:34:42 +0900] "GET /item/computers/2216 HTTP/1.1" 200 65 "/item/games/643" "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)" 40.66.156.206 - - [16/Jan/2018:18:34:42 +0900] "GET /item/games/4918 HTTP/1.1" 200 73 "/item/games/2145" "Mozilla/5.0 (iPad; CPU OS 5_0_1 like Mac OS X) AppleWebKit/534.46 (KHTML, like Gecko) Version/5.1 Mobile/9A405 Safari/7534.48.3" 48.201.76.35 - - [16/Jan/2018:18:34:42 +0900] "GET /category/toys HTTP/1.1" 200 85 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.1; WOW64; Trident/4.0; YTB730; GTB7.2; SLCC2; .NET CLR 2.0.50727; .NET CLR 3.5.30729; .NET CLR 3.0.30729; .NET4.0C; .NET4.0E; Media Center PC 6.0)"
②Crawler設定、実行
①で作成したS3用のクローラーを設定して実行します。
③Job作成、実行
左ペインのJobsを選択し「Add Job」をクリックしてJobを追加します。
この時、Job BookmarkをEnableに変更します。それ以外については環境に応じて設定してください。
Job bookmarkについては、以下の設定値があります。
デフォルトはDisableになります。
Job bookmark | 説明 |
---|---|
Enable | 前回実行した以降のデータを処理 |
Disable | 常に全データを処理 |
Pause | 前回実行した以降のデータを処理するが、状態を更新しない |
※Pauseはどういう時に使えばいいかが分かっていません。
Jobの設定が完了したら、実行してください。
④変換データ確認
上記でJobが正常に実行されたら、変換後のデータを確認します。
まずは、変換先のS3用のクローラーを設定します。
以下のようにクロール対象から_common_metadataと_metadataを除外するようにセットします。
クローラーを実行後、AthenaからSQLを実行してデータが変換されていることを確認します。
今回は1つ目のファイルにはアクセスログを10行用意しました。
⑤S3に追加のログをアップロード
準備で作成したapacheログファイルの二つ目をソースとなるS3にアップロードします。
⑥Job再実行
再度、同一Jobを実行します。
⑦変換データ確認
上記Jobが正常終了したら、Athenaから④で確認したの同様のSQLを実行して増分のみ追加されていることを確認してください。Job BookmarkがDisableの場合は、再実行時に全件処理されてしまいます。
もし想定通りの結果とならないようであればJob Bookmarkの設定値を再確認してください。
2つ目のファイルにはアクセスログを11行用意しているので、以下のとおり全部で21行となり増分データのみ追加されました
現状、このJob Bookmarkについては、ソースがS3のときのみ利用可能となっています。
RDSなどのDBの場合については、処理のなかでフィルタする必要があります。
実際にBookmarkを本番で使うには、ジョブがエラーとなった場合などどのようにリトライするかをしっかり想定しておく必要がありそうですね。
AWS Glueの開発環境をDockerで構築する(その2)
前回、Glueの開発環境をDockerで作りましたが、zeppelinコンテナを起動してホストからsshフォワードを実行して接続するという構成でホストを意識する必要がありましたが、せっかくzeppelinをdockerで作っているのでsshフォワードを行う部分もdockerで実現できないかと思い考えてみました。
今回作った構成
データを永続化するためにdataコンテナとsshフォワード用にssh forwardコンテナを追加することにしました。
そのためzeppelinコンテナからはssh forwardコンテナに接続しにいくことになります。
こうすることでssh forwardコンテナには名前でアクセスできるので、毎回設定変更する必要がなくなります。
手順
- ssh forwardコンテナ作成
- docker-compose.yml作成
- 起動確認
ssh forwardコンテナ作成
以下のとおりDockerfileを作成します。
FROM ubuntu:latest RUN apt-get -y update && apt-get -y upgrade RUN apt-get -y install openssh-client net-tools ADD ssh_forward.sh /home/ubuntu/ssh_forward.sh RUN chmod 644 /home/ubuntu/ssh_forward.sh CMD ["/bin/sh","/home/ubuntu/ssh_forward.sh"]
コンテナ起動時に実行するssh_forward.shを作成します。
起動時に環境変数で秘密鍵と接続先のIP or DNSを受取り鍵ファイルを作成して開発エンドポイントへ接続します。
#! /bin/sh echo $SSH_RSA > dev_endpoint.pem chmod 600 dev_endpoint.pem ssh -g -i dev_endpoint.pem -NTL 9007:169.254.76.1:9007 glue@$HOST_ADDRESS -o StrictHostKeyChecking=no
ビルド実行
docker build -t ssh_forward .
docker-compose.ymlを作成
version: '2' services: data: image: busybox container_name: app volumes: - /zeppelin/notebook zeppelin: image: apache/zeppelin:0.7.3 ports: - "8080:8080" volumes_from: - data restart: always links: - "ssh_server" ssh_server: image: ssh_forward environment: SSH_RSA: "[DevEndpoint接続用ssh秘密鍵]" HOST_ADDRESS: [GlueのDevEndpointのDNS or IP]
起動確認
busybox、apache/zeppellinについては予めdocker hubから取得しておいてください。
また、GlueのDevEndpointの設定も事前に必要になります。
docker-compose up
以下アドレスにアクセスして画面が表示されていれば成功です。
http://localhost:8080
これで全てdockerコンテナ上で完結できましたね。
AWS GlueでVPCフローログ用のclassifiersを作ってみた
GlueでVPCフローログをparquet形式に変換させる定期ジョブを作ろうと思いクロール処理を追加したところ、ビルトインのClassifiersにはなかったため自動でテーブル構造を認識してくれませんでした。
認識させるためにはカスタムClassifiersを作る必要があることが分かりました。
今回はVPCフローログ用にカスタムClassifiersの作り方を調べた備忘録です
GlueのビルトインのClassifiersについては、以下のページで確認できます。
ここになければ作る必要があります。
Adding Classifiers to a Crawler - AWS Glue
手順
- Grokフィルタ作成
- classifiersの作成
- 実行確認
Grokフィルタ作成
まずログをマッチングして分類するためのフィルタを作る必要がりあります。
Grokフィルタの詳細についてはここを確認してください。
Grok filter plugin | Logstash Reference [6.1] | Elastic
また、Glueのビルトインパターンについては以下を参照してください。
Writing Custom Classifiers - AWS Glue
作ったGrokフィルタをcrawlerを使って逐一実行確認していはかなりの時間がかかるため、
以下のサービスを使ってGrokフィルタが想定どおりに動作しているかを事前に確認します。
Grok Debugger
カスタムパターンを作る時の正規表現の確認は、以下のサービスが便利でした
Online regex tester and debugger: PHP, PCRE, Python, Golang and JavaScript
classifiersの作成
左ペインから「classifiers」を選択し、「Add classifier」ボタンをクリックします。
今回は以下のように入力してCreateします。
classifier name:任意
classifier type : Grok
classification:任意
Grok pattern:
%{INT:version:int} %{NOTSPACE:account_id} %{ENI:eni_id} %{IP:srcaddr} %{IP:dstaddr} %{INT:srcport:int} %{INT:dstport:int} %{INT:protocol:int} %{INT:packets:int} %{NUMBER:bytes} %{NUMBER:start} %{NUMBER:end} %{NOTSPACE:action} %{NOTSPACE:log_status}
Custome pattern:
ENI [eni-]+[a-z0-9]{8,8}
実行確認
上記設定ができれば実際にクロールしてみましょう。
Add crawlerを実行すると以下のような画面になります。
左下のclassifierリストから今回作ったclassifierをAddします。
vpcフローログのあるS3パスを指定して、それ以外は画面に沿って入力しRun crawlerを実行してください。
正しく分類できていれば、テーブルの詳細を確認すると以下のようになっていると思います。
うまくいかなかった場合は、ClassificationにUnknownと表示されますのでその場合は修正して再度確認してください。
ただし、classifiersを更新した際は、crawlerを新規に作成しないと反映されないようなので注意してください。
grokフィルタについては、少しクセがあり慣れるのに少し時間がかかりそうですが、Debuggerもあるのでうまく活用して慣れていきたいと思います。
AWS Glueで新しくScalaがサポートされました
AWS GlueのETLスクリプトを作成する言語として、新たにScalaが追加されました。
画面を確認すると以下のようにPythonに加えてScalaも選択できるようになっています。
以下はScalaで自動生成されたETLスクリプトになります。
import com.amazonaws.services.glue.ChoiceOption import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.MappingSpec import com.amazonaws.services.glue.ResolveSpec import com.amazonaws.services.glue.errors.CallSite import com.amazonaws.services.glue.util.GlueArgParser import com.amazonaws.services.glue.util.Job import com.amazonaws.services.glue.util.JsonOptions import org.apache.spark.SparkContext import scala.collection.JavaConverters._ object GlueApp { def main(sysArgs: Array[String]) { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) // @params: [JOB_NAME] val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) // @type: DataSource // @args: [database = "sampledb", table_name = "vpc_flow_logs", transformation_ctx = "datasource0"] // @return: datasource0 // @inputs: [] val datasource0 = glueContext.getCatalogSource(database = "sampledb", tableName = "vpc_flow_logs", redshiftTmpDir = "", transformationContext = "datasource0").getDynamicFrame() // @type: ApplyMapping // @args: [mapping = [("version", "int", "version", "int"), ("account", "string", "account", "string"), ("interfaceid", "string", "interfaceid", "string"), ("sourceaddress", "string", "sourceaddress", "string"), ("destinationaddress", "string", "destinationaddress", "string"), ("sourceport", "int", "sourceport", "int"), ("destinationport", "int", "destinationport", "int"), ("protocol", "int", "protocol", "int"), ("packets", "int", "packets", "int"), ("bytes", "int", "bytes", "int"), ("starttime", "int", "starttime", "int"), ("endtime", "int", "endtime", "int"), ("action", "string", "action", "string"), ("logstatus", "string", "logstatus", "string")], transformation_ctx = "applymapping1"] // @return: applymapping1 // @inputs: [frame = datasource0] val applymapping1 = datasource0.applyMapping(mappings = Seq(("version", "int", "version", "int"), ("account", "string", "account", "string"), ("interfaceid", "string", "interfaceid", "string"), ("sourceaddress", "string", "sourceaddress", "string"), ("destinationaddress", "string", "destinationaddress", "string"), ("sourceport", "int", "sourceport", "int"), ("destinationport", "int", "destinationport", "int"), ("protocol", "int", "protocol", "int"), ("packets", "int", "packets", "int"), ("bytes", "int", "bytes", "int"), ("starttime", "int", "starttime", "int"), ("endtime", "int", "endtime", "int"), ("action", "string", "action", "string"), ("logstatus", "string", "logstatus", "string")), caseSensitive = false, transformationContext = "applymapping1") // @type: ResolveChoice // @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"] // @return: resolvechoice2 // @inputs: [frame = applymapping1] val resolvechoice2 = applymapping1.resolveChoice(choiceOption = Some(ChoiceOption("make_struct")), transformationContext = "resolvechoice2") // @type: DropNullFields // @args: [transformation_ctx = "dropnullfields3"] // @return: dropnullfields3 // @inputs: [frame = resolvechoice2] val dropnullfields3 = resolvechoice2.dropNulls(transformationContext = "dropnullfields3") // @type: DataSink // @args: [connection_type = "s3", connection_options = {"path": "s3://glue-test-out-bucket"}, format = "parquet", transformation_ctx = "datasink4"] // @return: datasink4 // @inputs: [frame = dropnullfields3] val datasink4 = glueContext.getSinkWithFormat(connectionType = "s3", options = JsonOptions("""{"path": "s3://glue-test-out-bucket"}"""), transformationContext = "datasink4", format = "parquet").writeDynamicFrame(dropnullfields3) Job.commit() } }
Sparkの利用者はPythonかScalaのどちらの言語の利用が多いんでしょうか?
いずれにせよ言語の選択肢が増えるのはすごくいいですね。
AWS Glueの開発環境(Zeppelin)をDockerで構築する
AWS Glueで自動生成されたETL処理のPySparkの開発について、AWSコンソール上で修正して実行確認は可能ですがかなり手間になります。
そこで開発エンドポイントを使って開発する方法が提供されており、Apache Zeppelinなどを使ってインタラクティブに開発することができます。公式ドキュメントによると、エンドポイントを利用して開発する方法としては大きく以下の4つの方法が提供されています。
① EC2上にApache Zeppelinを構築して開発エンドポイントへ接続
② ローカルマシンにApache Zeppelinを構築して開発エンドポイントへ接続
③ 開発エンドポイントへ直接sshしてREPL Shellを利用する
④ PyCharmのProfessional editionで開発エンドポイントへ接続
①の方法がコンソールからワンクリックで開発環境を構築することができるので簡単かつ便利に準備することができますが、EC2(m4.xlarge)が起動されるためかなりの費用がかかります。
今回は②の方法でかつローカルマシン上のDockerでApacheZeppelinを構築して開発環境を準備してみたいと思います。
1. 開発エンドポイントを作成
事前に以下が対応されている前提とします
詳細は以下参照
docs.aws.amazon.com
開発エンドポイントについては、費用1DPUあたり0.44ドル/時で最低2DPUからとなりますので
できるだけこまめに消すことを想定してCloudFormationで作成します。
AWSTemplateFormatVersion: '2010-09-09' Description: "Glue DevEndpoint" Resources: DevEndpoint: Type: "AWS::Glue::DevEndpoint" Properties: EndpointName: "TestDevEndpoint" NumberOfNodes: 2 PublicKey: "[ssh public key]" RoleArn: "[Glue Service RoleのARN]" SecurityGroupIds: - [開発エンドポイント用のセキュリティグループ] SubnetId: [開発エンドポイント用のサブネット]
上記を作成して保存後、コンソールもしくはcliから実行してください。
aws cloudformation create-stack --stack-name stack-dev-endpoint --region ap-northeast-1 --template-body file://dev_endpoint.yml
と思ったら、エラーが発生してうまく実行できませんでいた。
バージニアリージョンでは問題なく実行できましたので、おそらく現時点で東京リージョンではGlueのCloudFormationがまだ使えないようです。
仕方が無いのでcliで作成することにします。
以下のコマンドを実行して開発エンドポイントを作成します。
aws glue create-dev-endpoint \ --endpoint-name TestDevEndpoint \ --role-arn [Glue Service RoleのARN] \ --security-group-ids sg-866cdeff \ --subnet-id [開発エンドポイント用のセキュリティグループ] \ --public-key "[ssh public key]" \ --number-of-nodes 2
実行後しばらくしてstatusがREADYになれば準備完了です。
3. 開発エンドポイントにsshフォワードで接続
コンソールから開発エンドポイントの詳細を開き「SSH tunnel to remote interpreter」に記載のある接続コマンドをコピーします。
コピーしたコマンドに-gオプションを付与してMacからssh接続を行います
ssh -g -i <private-key.pem> -NTL 9007:169.254.76.1:9007 glue@ec2-xxx-xxx-xxx-xxx.ap-northeast-1.compute.amazonaws.com
※コンテナからトンネルを利用できるようにするためにgオプションを付与します
4. Apache/Zeppelinを起動&設定
docker run -p 8080:8080 apache/zeppelin:0.7.3
※タグは取得したバージョンをセットしてください
起動後以下のURLにアクセスします
http://localhost:8080.
画面右側からinterpreterを選択します
spark項目で以下のように設定します。
- Connect to existing processをチェック
- HostにローカルマシンのIP
- masterにyarn-clinet
- psark.executor.memoryを削除
設定後にrestartすれば完了です。
5. 動作確認
以下のサンプルコードを実行してスキーマ情報が出力されることを確認してください
DataBase、TableNameは実際に設定しているものを使用してください
%pyspark import sys from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.transforms import * # Create a Glue context glueContext = GlueContext(SparkContext.getOrCreate()) # Create a DynamicFrame using the 'persons_json' table persons_DyF = glueContext.create_dynamic_frame.from_catalog(database="legislators", table_name="persons_json") # Print out information about this data print "Count: ", persons_DyF.count() persons_DyF.printSchema()
これで気軽に試せる環境が整いました。
実際に使うにはコードのデータや設定データの永続化も考える必要がありますが、まずはこれで色々と試してみてください。
DockerではじめるPySpark
最近Sparkの勉強を始めました。
手軽に試せる環境としてPySparkをJupyter Notebookで実行できる環境を作ればよさそうです。
環境構築に手間取りたくなかったので、Dockerで構築できないか調べてみるとDocker Hubでイメージが提供されていましたので、それを利用することにしました。
今回は導入からサンプル実行までやってみたいと思います。
導入手順
Dockerイメージの取得
以下のコマンドを実行してイメージを取得してください。
イメージのサイズが約5GBあるのでディスク容量には気をつけてください
docker pull jupyter/pyspark-notebook
Dockerの起動
notebookのデータを保存しておくために、ローカルに適当なディレクトリを作成してDockerのマウント先を作ります。
以下コマンドを実行してDockerを起動します。
docker run -p 8888:8888 -v ローカルボリュームのフルパス:/home/jovyan/work jupyter/pyspark-notebook start-notebook.sh --NotebookApp.token=''
Jupyterの起動
Dockerの起動完了後、以下のアドレスにアクセスします
http://localhost:8888/
正常に起動していれば以下のような画面が表示されます
右端のNewボタンからPython3を選択します
新規に開いた画面に以下のコードを入力します
from pyspark.context import SparkContext sc = SparkContext() data = sc.parallelize([1, 2, 3, 4, 5]) print(data.count()) sc.stop()
Runボタンをクリックし実行結果が表示されれば成功です
後は色々とサンプルを動かしてみてください。
Dockerで手軽に試せるのはいいですね。