HadoopからSparkへの移行

こんにちは。配信/インフラチームの佐々木です。今回は、adstirの集計システムをHadoopからSparkに移行したお話をしたいと思います。

以前の構成

adstirでは以前はAWS上のEMRでHadoopを使って集計していました。一つの集計のフローとしては以下のようになります。

Mapper(UNIQ集計)
↓
SQSにメッセージを送信
↓
Reducer(UNIQ集計)
↓
SQSにメッセージを送信
↓
Mapper(通常集計)
↓
SQSにメッセージを送信
↓
Reducer(通常集計)

このシステムでは以下のような問題点があり、それを解消するために移行を計画しました。

・Hadoopのパフォーマンスがあまり良くなく、集計に時間がかかり、かつ料金的なコストもかかる。

・MapperとReducerでアプリケーションがを分ける必要があり、UNIQ集計も別で必要になるため、煩雑な仕組みになる。

・かなり昔に開発したシステムの為、利用しているサービスや構成が古く、パフォーマンス・安定性等お世辞にも良いと言えない為、最新のアーキテクチャに刷新したい。

Sparkに移行すれば、UNIQ集計を分ける必要もなく、SQSでの通知も不要になります。つまり上記のフローが一つの処理で完結するわけで、システムが大幅に簡素化し、かつ時間短縮・コスト削減するのを見込み開発をしました。

移行する集計用のソフトウェアの検討

当初はAthenaに移行する想定でしたが、Athenaは分析用のサービスで、今回のようなバッチでの集計処理には向いていないことがわかり、Hadoopから移行しやすくかつパフォーマンスの良いSparkを採用することにしました。ですが手動での分析ではAthenaを使いたいとの要望があり、バッチ集計をSpark、分析用にAthenaを使い分けをするようにしました。

Sparkについて

Sparkとは一言で言えば「オンメモリで動くHadoop」の事で、AWSではEMRでのマネージドサービスを使うことが出来ます。利用できる言語はScala、Python、R、Java等ですが一般的にはScalaとPythonを使うことが多いようです。Spark自体はScalaで実装されています。

採用した言語とAPI

当初はPythonで実装したのですが、テストしてみたところHadoopとあまりパフォーマンスが変わりませんでした。原因はRDDでの処理がボトルネックになっている事でした。 Sparkで使える用意されているAPIはRDD、Dataframe、Datasetの3種類があります。APIの特徴は以下のページに詳しく書かれています。

https://yubessy.hatenablog.com/entry/2016/12/11/095915 https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html

一見PythonでDataframeを使うのが良さそうなのですが、Dataframeは色々制限がある為、複雑なデータ構造を扱う場合はRDDの方が向いています。そこで全てScalaに書き換え、APIにRDDとDataframeを併用する実装にしました。またSparkはExecuter（分散）とDriver（非分散）で処理が分かれており、できる限りExecuter側で処理をすればパフォーマンスが向上しますので、そこを意識して実装する事が大事でした。

結果

集計時間とコストが1/3程度になり、期待する結果を得る事が出来ました。 AWSでは集計用のマネージドサービスが他にも多々あり、これからも追加されていくと思われますので、今後もキャッチアップして行きたいと思います。

今回は以上となります。

EC2上のMySQLからRDSへの移行

こんにちは。配信/インフラチームの佐々木です。

少し前の話になりますが、Bypass(DSP)のRDBをEC2上のMySQLからからRDS(Aurora)に移行しました。データ容量が膨大なためmysqldumpを利用した移行は出来ず、色々苦労しましたのでそう言った点を記載します。

移行時のレプリケーション構成

移行するにあたり、事前に本番のデータをRDSに移しレプリケーションさせておく必要があるのですが、移行対象のMySQLはステートメントベースのレプリケーションのため、構成としては以下のようになります。

本番のMySQL(EC2)
↓ステートメントベースでレプリケーション
レプリケーション用のMySQL(EC2)
↓行ベースでレプリケーション
Aurora

AMIイメージを利用したMySQLのレプリケーション

前述の通りEC2 to EC2のレプリケーションをmysqldumpでは出来ないため、AMIイメージでMySQLをインスタンス丸ごとコピーしレプリケーションすることで実現しました。あまり一般的なやり方では無いのですが、イメージを取得する直前（バイナリログの最後のポジション）のポジションを探し出し、レプリケーション開始時にそのポジションを指定することで実現しました。

RDSへのレプリケーション

EC2からRDSにレプリケーションする際はDMS(Database Migration Service)を利用するのが一般的と思います。
ですがDMSを利用してのレプリケーションが上手くいかなかったため、XtraBackupというMySQL用のバックアップツールを利用してS3上にデータをバックアップし、そのデータをRDSにリストア後、レプリケーションを開始することで実現しました。このツールはバックアップ時にバイナリログのポジションもinfoとして残るため、バイナリログの差分を探す必要もありません。XtraBackup はDMSと違いCLIでの設定になりますが、使ってみた感じでは特に難しいところはなく、バックアップ時間もmysqldumpに比べると大幅に削減できますので、大容量のDBのバックアップ時にはオススメ出来るツールです。

Auroraの設定

RDSの仕様として文字コードが一部変更されてしまっていたので、クラスターのパラメータグループで設定し直しました。
参考:https://qiita.com/reoy/items/e355debf1e2b2abd703b
またユーザ関連の設定も一部差分が出ましたが、大きな問題はありませんでした。

今回は以上となります。

DynamoDBストリームの利用

こんにちは。技術開発部・配信/インフラチームの二階堂です。弊社DSP「Bypass」ではDynamoDBを利用しております。今回はその機能の一つであるDynamoDBストリームを紹介したいと思います。

DynamoDBストリームとは

DynamoDBストリームはDynamoDBへの書き込み・更新・削除処理(ttlによる自動削除を含む)を最大24時間保存する機能です。テーブル設定で「ストリーム有効」にすると保存されるようになります。保存内容は「キーのみ」「新しいイメージ」「古いイメージ」「新旧イメージ」の4種類から選択可能で、必要な情報のみを保存することで使用料を抑えることが可能です。

ストリームレコードの保存形式

DynamoDBストリームに保存されたデータを扱うにあたり、まずはデータがどのような形式で保存されるかを説明したいと思います。低レベルAPIを利用する場合はこの知識が必須です。

DynamoDBストリームのデータは下図のように複数のシャードから構成され、シャードは複数のストリームレコードから構成されます。ストリームレコードはDynamoDBへの書き込み処理などリクエスト一つ一つに対応します。シャードには順番が存在し各シャードには次のシャードのidが保存されています。

(図は公式ページより引用)

ストリームレコードの取得方法

低レベルAPIを利用してデータを取得する際には次の手順で取得します。

DynamoDBのテーブル名からdescribe-tableでLatestStreamArnを取得する
DynamoDBStreamsのdescribe-streamで取得したLatestStreamArnからシャードのリストを取得する
リストの一番最初のシャードのシャードidとLatestStreamArnを使ってget-shard-iteratorでイテレータを取得する
取得したイテレータを使ってget-recordsでストリームレコードのリストを取得する
ストリームレコードを各々処理する
2で取得したシャードそれぞれに対して3~5を行う

この手順の注意点は2,4でリストを取得する時に対象が多すぎると一度に全て取得することができない点です。手順2ではLastEvaluatedShardIdが入っていた場合、手順4ではNextShardIteratorが入っていた場合にはその値を使って再度取得する必要があります。

また、手順4ではシャードが閉じていない場合には次のストリームレコードが存在しなくてもNextShardIteratorが入っています。そのため常時実行するタイプの処理ではない場合には、数回連続でget-recordsの結果が空だった場合にはNextShardIteratorが入っていても中断するなどの処理が必要になります。

まとめ

DynamoDBストリームはttlによる削除検知など痒いところに手が届く便利な機能ですが取得方法が複雑で躓きやすい部分でもあります。この記事にがその一助となれば幸いです。

今回は以上となります。

決定木アンサンブルモデルのデプロイを容易にするライブラリtreeliteの紹介(2)

技術開発部の安井です。前回に引き続き、treeliteの紹介をしたいと思います。今回はtreeliteの予測速度について検証しようと思います。ドキュメント上では2~6倍のスループットが出ると書かれています。この点に関して実際にpython上とgolang上の2つの言語環境で検証しようと思います。

使用データと学習について

今回予測速度を検証する際に使用するデータはAvazuのclick予測のデータセットを使用します。全データ使用すると学習に時間がかかるので、300万件のデータのみを使用して学習を行います。Avazuのデータセットは変数がすべてカテゴリカル変数になっています。今回はこれらの変数をFeeatureHasherを使用して特徴量に変換します。今回FeatureHasherを使用する理由は後述します。FeatureHasherの次元数として今回は16、64、256次元でそれぞれ検証を行います。学習には前回に引き続きLightGBMを使用し、num_boost_roundを100で設定します。これで弱学習器として決定木が100本生成されることになります。学習するときに使用したコードを以下に記述します。

#! /usr/bin/env python
    
import sys
import pandas as pd
import lightgbm as lgb
import treelite
import treelite.runtime
from sklearn.feature_extraction import FeatureHasher
from sklearn.model_selection import train_test_split

train = pd.read_csv('input/train', nrows=4000000)
    
dim = int(sys.argv[1])
fh = FeatureHasher(dim)
X = fh.transform(train.drop(['id', 'click'], axis=1).to_dict('records'))
y = train['click']
    
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=47)
dtrain = lgb.Dataset(X_train, label=y_train)
dtest = lgb.Dataset(X_test, label=y_test)
    
params = {
    'objective': 'binary',
    'metric': 'auc',
    'boosting': 'gbdt',
    'learning_rate': 0.01,
    'seed': 47,
}
    
bst = lgb.train(
    params,
    dtrain,
    num_boost_round=100,
    valid_sets=[dtrain, dtest],
    valid_names=['train', 'test'],
)
lgb_preds = bst.predict(X_test)
    
bst.save_model(f'fh_lgb_dim_{dim}.model')
    
model = treelite.Model.load(f'fh_lgb_dim_{dim}.model', model_format='lightgbm')
model.export_lib(toolchain='gcc', libpath=f'./fh_lgb_dim_{dim}.so', verbose=True)
    
pd.DataFrame(X_test.toarray(), columns=['var_{}'.format(i + 1) for i in range(dim)]).to_csv(f'X_test_fh_dim_{dim}.csv', index=False)

#! /usr/bin/env python

import sys

import pandas as pd

import lightgbm as lgb

import treelite

import treelite.runtime

from sklearn.feature_extraction import FeatureHasher

from sklearn.model_selection import train_test_split

train = pd.read_csv('input/train', nrows=4000000)

dim = int(sys.argv[1])

fh = FeatureHasher(dim)

X = fh.transform(train.drop(['id', 'click'], axis=1).to_dict('records'))

y = train['click']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=47)

dtrain = lgb.Dataset(X_train, label=y_train)

dtest = lgb.Dataset(X_test, label=y_test)

params = {

'objective': 'binary',

'metric': 'auc',

'boosting': 'gbdt',

'learning_rate': 0.01,

'seed': 47,

}

bst = lgb.train(

params,

dtrain,

num_boost_round=100,

valid_sets=[dtrain, dtest],

valid_names=['train', 'test'],

)

lgb_preds = bst.predict(X_test)

bst.save_model(f'fh_lgb_dim_{dim}.model')

model = treelite.Model.load(f'fh_lgb_dim_{dim}.model', model_format='lightgbm')

model.export_lib(toolchain='gcc', libpath=f'./fh_lgb_dim_{dim}.so', verbose=True)

pd.DataFrame(X_test.toarray(), columns=['var_{}'.format(i + 1) for i in range(dim)]).to_csv(f'X_test_fh_dim_{dim}.csv', index=False)

pythonでの予測速度比較

まずはじめにpythonでの予測速度の比較を行います。予測速度は以下の3パターンのデータ量で比較を行います。

instance: 1件
batch: 1,000件
big batch: 1,000,000件

予測速度は複数回予測したときの平均時間を記載しています。

16次元の場合

lightgbm instance:  0.00010517575740814209s
treelite instance:  0.00003566386699676514s
lightgbm batch:     0.0007027374267578125s
treelite batch:     0.0003089529752731323s
lightgbm big batch: 0.6247398257255554s
treelite big batch: 0.2774349188804626s

lightgbm instance: 0.00010517575740814209s

treelite instance: 0.00003566386699676514s

lightgbm batch: 0.0007027374267578125s

treelite batch: 0.0003089529752731323s

lightgbm big batch: 0.6247398257255554s

treelite big batch: 0.2774349188804626s

16次元の場合、treeliteの方がおおよそ1/2~1/3の時間で予測ができており、概ねドキュメントの記載の通りであることが見受けられます。

64次元の場合

lightgbm instance:  0.0001033491849899292s
treelite instance:  0.00009161360263824462s
lightgbm batch:     0.0007573297739028931s
treelite batch:     0.00035665130615234375s
lightgbm big batch: 0.79298579454422s
treelite big batch: 0.32156704187393187s

lightgbm instance: 0.0001033491849899292s

treelite instance: 0.00009161360263824462s

lightgbm batch: 0.0007573297739028931s

treelite batch: 0.00035665130615234375s

lightgbm big batch: 0.79298579454422s

treelite big batch: 0.32156704187393187s

64次元の場合、バッチ予測の場合は16次元と同様にtreeliteの方がおおよそ1/2~1/3の時間で予測ができています。しかし、instance予測の場合lightgbmとtreeliteの予測速度はそこまで差がないように見受けられます。

256次元の場合

lightgbm instance:  0.00010415022373199463s
treelite instance:  0.0003009482145309448s
lightgbm batch:     0.0011013625144958496s
treelite batch:     0.00042464187145233154s
lightgbm big batch: 1.5125486660003662s
treelite big batch: 0.38220595598220825s

lightgbm instance: 0.00010415022373199463s

treelite instance: 0.0003009482145309448s

lightgbm batch: 0.0011013625144958496s

treelite batch: 0.00042464187145233154s

lightgbm big batch: 1.5125486660003662s

treelite big batch: 0.38220595598220825s

256次元の場合、バッチ予測の場合はtreeliteの方が1/3~1/4の時間で予測ができており、treeliteの効果が最も出ていると考えられます。対象的に、instance予測の場合はtreeliteの方が予測に3倍の時間がかがっていることがわかります。

まとめ

バッチ予測の場合は概ね期待している効果が出ていますが、インスタンス予測の場合は予測する特徴量のサイズが増えるほどの効果が出ないばかりか、悪化する場合も存在します。これに関してはtreelite側のメソッドがバッチ予測の場合とインスタンス予測の場合で異なることが原因として考えられます。

バッチ予測: treelite.Predictor.predict(batch, verbose=False, pred_margin=False)
インスタンス予測: treelite.Predictor.predict_instance(inst, missing=None, pred_margin=False)

この2つの違いに関してですが、batchの方はtreelite側で処理をしやすいようなデータ型を期待していますが、instの方ではnumpy.ndarrayやscipy.sparse.csr_matrix等のarray objectを期待しています。この差により予測速度に差が出ているのではないかと想像していますが、詳しくは引き続き調査を続けていきたいと思います。

golangでの予測速度比較

golangでの予測速度比較にはtreeliteで生成したモデルコードの予測と以下のGBDT予測ライブラリの予測での速度比較を行います。

dmitryikh/leaves

このライブラリはXGBboost/LightGBm/scikit-learnで学習したGBDTのモデルをロードしてgolang上で予測を行えるようにしたライブラリです。こちらはインスタンス予測のみを比較することとします。以下がgolanで測定したベンチマークの結果です。

16次元の場合
$ go test -bench . -benchmem
goos: linux
goarch: amd64
pkg: treelite_bench/dim16
BenchmarkPredictTreelite-8       2000000               849 ns/op              64 B/op          1 allocs/op
BenchmarkPredictLeaves-8          500000              3655 ns/op               8 B/op          1 allocs/op
PASS
ok      treelite_bench/dim16       4.426s

64次元の場合
$ go test -bench . -benchmem
goos: linux
goarch: amd64
pkg: treelite_bench/dim64
BenchmarkPredictTreelite-8       2000000               637 ns/op             256 B/op          1 allocs/op
BenchmarkPredictLeaves-8          500000              3546 ns/op               8 B/op          1 allocs/op
PASS
ok      treelite_bench/dim64       3.717s

256次元の場合
$ go test -bench . -benchmem
goos: linux
goarch: amd64
pkg: treelite_bench/dim256
BenchmarkPredictTreelite-8       2000000               890 ns/op            1024 B/op          1 allocs/op
BenchmarkPredictLeaves-8          300000              4220 ns/op               8 B/op          1 allocs/op
PASS
ok      treelite_bench/dim256       3.981s

16次元の場合

$ go test -bench . -benchmem

goos: linux

goarch: amd64

pkg: treelite_bench/dim16

BenchmarkPredictTreelite-8 2000000 849 ns/op 64 B/op 1 allocs/op

BenchmarkPredictLeaves-8 500000 3655 ns/op 8 B/op 1 allocs/op

PASS

ok treelite_bench/dim16 4.426s

64次元の場合

$ go test -bench . -benchmem

goos: linux

goarch: amd64

pkg: treelite_bench/dim64

BenchmarkPredictTreelite-8 2000000 637 ns/op 256 B/op 1 allocs/op

BenchmarkPredictLeaves-8 500000 3546 ns/op 8 B/op 1 allocs/op

PASS

ok treelite_bench/dim64 3.717s

256次元の場合

$ go test -bench . -benchmem

goos: linux

goarch: amd64

pkg: treelite_bench/dim256

BenchmarkPredictTreelite-8 2000000 890 ns/op 1024 B/op 1 allocs/op

BenchmarkPredictLeaves-8 300000 4220 ns/op 8 B/op 1 allocs/op

PASS

ok treelite_bench/dim256 3.981s

BenchmarkPredictTreelite-8がtreeliteで予測した場合のベンチマークで、BenchmarkPredictLeaves-8がleavesで予測した場合のベンチマークです。 treeliteの予測モデルのほうが1/4~1/6の時間で予測が行えれていることがわかります。これを見ると動的にロードしたleavesの予測モデルより、はじめに実行バイナリに組みこんでおいたtreeliteの予測モデルのほうがパフォーマンスが良いことがわかります。

treelite使用上の注意

冒頭でも述べましたが今回はカテゴリカル変数をFeatureHasherを使用して学習するときに使用する特徴量を作成しました。しかし、LightGBMではカテゴリカル変数をそのまま特徴量として入力することができるようになっています。(厳密にはint型のidへの変換は必要) 今回カテゴリカル変数をそのまま学習に用いなかった理由としてはLightGBMでカテゴリカル変数を使用して学習を行うとtreeliteの予測結果と異なってしまうという事象が発生したためです。しかし、現在では以下のissueによってこの問題も解決されたようです。

https://github.com/dmlc/treelite/issues/77

実際手元の環境でカテゴリカル変数を用いた場合でもLightGBMの予測モデルとtreeliteの予測モデルの結果が一致することを確認しました。古いバージョンのtreeliteを使用する場合はご注意ください。

さいごに

今回はtreeliteの予測速度性能に関してpython, golang上で比較を行いました。 golangでの予測、pythonでのバッチ予測に関してはスピードアップしていることが確認できました。最近は予測に特化したライブラリが他にも出てきているので実環境に組み込むといった観点でこういった予測に特化したライブラリもチェックしていきたいと思います。

ClassicLinkを使用してVPCとEC2-Classic間の相互通信を行う

こんにちは。技術開発部部長の川住です。

弊社DSP「Bypass」はシステムをAWSに移行してから5年以上が経過しました。近年ではEC2インスタンスをVPC (Virtual Private Cloud) 内に作成し運用していますが、AWSへの移行当初にVPC外（EC2-Classic）に作成したサーバがいくつか現在も稼働しており、「VPC内で稼働しているインスタンス」と、「VPC外で稼働しているインスタンス」が混在している環境となっています。今回はこのような環境で使用すると便利な「ClassicLink」という機能を紹介します。

ClassicLinkを使うとできること

VPC外のインスタンスと特定のVPCの相互接続

ClassicLinkを使用すると、VPC内（外）のサーバがあたかも同一のLAN上に存在するかのように、VPC外（内）のインスタンスからアクセスできるようになります。これによって、VPC外インスタンスのVPC内への移行を段階的に行えますし、移行の際のダウンタイムを減らせます。

ClassicLinkではできないこと

異なるVPC間の相互接続

ClassicLinkでは、「VPC外 (EC2-Classic) 」のインスタンスと特定のVPCとの相互接続を行う機能であり、VPC間の相互接続には使用できません。このような場合には、「VPC peering」機能を使用して相互接続の設定を行う必要があります。

VPC外のインスタンスと、複数のVPCとの相互接続

ClassicLinkはサーバごとに相互接続を行うVPCを設定できますが、ClassicLinkで接続できるVPCは1インスタンスに対して1つだけです。複数のVPCへのアクセスを実現にするには「VPC peering」機能と、VPCのルーティング設定を組み合わせる必要があります。

ClassicLinkの使い方

ClassicLinkの設定はAWSのコンソール画面上、またはaws-cliで行えます。設定したいインスタンスを選択し、接続したいVPC名と、VPC外からの通信の際にVPC側で適用するセキュリティグループを選択します。VPC内のセキュリティグループからはVPC外のセキュリティグループは参照できないので、EC2-ClassicのIP帯などを指定したインバウンドルールを設定したセキュリティグループを作成しておく必要があります。

ClassicLink使用時の注意点

ClassicLinkは起動中のインスタンス対してのみ設定でき、停止すると設定が失われてしまうので注意が必要です。

まとめ

使用できる場面は少ないですが、ClassicLinkを使用することで、VPC内外の相互接続を簡単に行えて、それによってVPC内への移行作業の手間やダウンタイムの削減が可能です。

今回は以上となります。

決定木アンサンブルモデルのデプロイを容易にするライブラリtreeliteの紹介(1)

はじめまして。技術開発部の安井です。現在は弊社で運営しているDSPにおける広告効果の分析や機械学習モデルの構築/運用を行なっています。

DSPと機械学習

詳細なDSPに関する説明は省きますが、DSPではSSPから受け取った広告リクエストに対して以下の事項を決定して広告オークションにおける入札応答を行わなければいけません。

入札可能な広告の内、どの広告で入札を行うのか
いくらで入札を行うのか

入札する広告、金額を決定にはクリック率、コンバージョン率等の広告リクエストに対する実際のパフォーマンスの値をを活用しています。これらの広告パフォーマンスを入札サーバ上で予測するために弊社では機械学習モデルを活用しています。しかし、DSPにおける機械学習モデルの活用には以下のようなハードルが存在します。

入札サーバを構築しているプログラムと機械学習モデルを学習しているプログラムでは使用しているプログラミング言語が異なる
SSPへの入札応答にはシビアな時間制約が存在する

これらを解決するための方法として今回はtreeliteというライブラリをご紹介します。

treelite

treeliteとは近年巷でよく使われているXGBoostやLightGBM等のGBDTの実装ライブラリやscikit-learnに実装されているRandomForestRegressor(Classifier)、GradientBoostingRegressor(Classifier)等のいわゆる決定木をアンサンブルしているモデルのデプロイを容易にするためのライブラリになっています。加えて、モデルが直接Cコードとして吐き出されるため、推論時間の高速化も期待できる実装となっています。開発は以下のリポジトリで行われています。

dmlc/treelite

Referenceはこちら

Treelite: toolbox for decision tree deployment

treeliteを用いたモデルのデプロイのオプションとして次の3つのオプションが示されています。

1.デプロイターゲットのマシンにtreeliteをインストールする

こちらはデプロイ先のマシンにtreeliteをインストールしておくシンプルな方法です。ゆえにデプロイ先のマシンにpythonがインストールされている必要があります。

2.デプロイターゲットのマシンにランタイムパッケージと一緒にデプロイする

treeliteではランタイムパッケージをアウトプットすることができます。このランタイムパッケージをデプロイプロセスで同時にターゲットマシンにデプロイすることで、モデルを使用できるようにする方法です。こちらもデプロイ先のマシンにpythonがインストールされている必要があります。

3.モデルとなるCコードのみデプロイする

こちらはCコードとしてアウトプットされたモデルのみをターゲットマシンに送る方法です。 Cコードに変換されたモデルは以下のインタフェースから予測値を取得できます。

float predict(union Entry* data, int pred_margin);

1	float predict(union Entry* data, int pred_margin);

dataが予測を行うときに必要となる特徴量の集合で、pred_marginはモデルがアウトプットした値をシグモイド関数に通すか通さないかというフラグとなっています(2値分類タスクの場合)。union Entryに関しては以下のように定義されており、特徴量が存在しない場合はmissingに-1が設定されており、特徴量が存在する場合はfvalueに値が格納されている状態を定義します。qvalueに関しては現状使われていないように見受けられます。

union Entry {
  int missing;
  float fvalue;
  int qvalue;
};

union Entry {

int missing;

float fvalue;

int qvalue;

};

golangでのtreeliteモデルの使用方法

上記3種類のデプロイ方法のうち、3番目のCコードのみをデプロイする方法であればpredict関数とのつなぎ込みの部分さえ実装すればpython以外の言語でもアンサンブルモデルを使用できます。弊社の入札サーバではgolangが採用されており、cgoを使うことでアンサンブルモデルのつなぎ込みを実現することができます。今回はLightGBMのモデルをtreeliteのモデルに変換し、golangから参照する部分までご紹介します。まずは、LightGBMのモデルを読み込みtreeliteのモデルをアウトプットする部分です。LightGBMのモデルはlgb.modelというファイルに保存されていることとします。

#! /usr/bin/env python
    
import treelite
    
model = treelite.Model.load('lgb.model', model_format='lightgbm')
model.export_lib(toolchain='gcc', libpath='./lgb.so', verbose=True)
model.export_srcpkg(platform='unix', toolchain='gcc',
                    pkgpath='./lgb.zip', libname='lgb.so',
                    verbose=True)

#! /usr/bin/env python

import treelite

model = treelite.Model.load('lgb.model', model_format='lightgbm')

model.export_lib(toolchain='gcc', libpath='./lgb.so', verbose=True)

model.export_srcpkg(platform='unix', toolchain='gcc',

pkgpath='./lgb.zip', libname='lgb.so',

verbose=True)

5行目のloadでLightGBMのモデルを読み込み6行目で共有ライブラリに変換を行います。その後、7行目で変換した共有ライブラリから必要なヘッダーとCコード、メタ情報を抽出し、zipファイルにまとめます。デフォルトだとzipファイルには以下のファイルがまとめられています。

header.h
main.c
Makefile
recipe.json

header.h

main.c

Makefile

recipe.json

header.hとmain.cが実際のモデル部分でそれらをコンパイルフローがMakefileにまとまっています。recipe.jsonにはモデルコードのメタ情報が格納されています。それではこれらのモデルをgolangから参照できるようにしましょう。コードはこちらとなります。

package main

//#cgo LDFLAGS: -lm
//#include "header.h"
import "C"
import (
    "encoding/binary"
    "fmt"
    "github.com/dmitryikh/leaves"
    "math"
    "unsafe"
)

func float32ToByteArray(flt float32) [4]byte {
    b := make([]byte, 4)
    binary.LittleEndian.PutUint32(b, math.Float32bits(flt))
    return *(*[4]byte)(unsafe.Pointer(&b[0]))
}

func predictTreelite(fvals []float32) float32 {
    cfvals := [16][4]byte{}
    for i, fval := range fvals {
        cfvals[i] = float32ToByteArray(fval)
    }
    p := C.predict(&cfvals[0], C.int(0))
    return float32(p)
}

func main() {
    fvals32 := make([]float32, 16)
    fmt.Println(predictTreelite(fvals32))
}

package main

//#cgo LDFLAGS: -lm

//#include "header.h"

import "C"

import (

"encoding/binary"

"fmt"

"github.com/dmitryikh/leaves"

"math"

"unsafe"

)

func float32ToByteArray(flt float32) [4]byte {

b := make([]byte, 4)

binary.LittleEndian.PutUint32(b, math.Float32bits(flt))

return *(*[4]byte)(unsafe.Pointer(&b[0]))

}

func predictTreelite(fvals []float32) float32 {

cfvals := [16][4]byte{}

for i, fval := range fvals {

cfvals[i] = float32ToByteArray(fval)

}

p := C.predict(&cfvals[0], C.int(0))

return float32(p)

}

func main() {

fvals32 := make([]float32, 16)

fmt.Println(predictTreelite(fvals32))

}

ディレクトリ構成は以下のようになっていることを想定しており、こちらのコードはmain.goに記述されています。

.
├── main.go
├── header.h
└── main.c

├── main.go

├── header.h

└── main.c

こちらのコードは16次元のベクトルを受け取り0~1までの確率値を返すモデルを想定しています。予測を行う関数はfunc predictTreelite(fvals []float32) float32ですがまず最初に受け取ったベクトルをバイナリ列の配列に変換しています。これはunion Entryを表現するための型がgolang側に存在しないためこのような形を取っています。今回はベクトルの中にmissingが存在しない想定で実装がされていますが、missingが存在しうる場合は-1のバイナリ列を代わりに入れてあげるようにします。このようにしてバイナリ列の配列に変換されたベクトルをモデルファイルのpredictに渡してあげることでモデルの予測値を獲得します。実際にビルドして実行すると以下のように0~1までの確率値が出力されます。

$ ./predict
0.3820182

1 2	$ ./predict 0.3820182

さいごに

今回はtreeliteでLightGBMで学習されたモデルをtreeliteのモデルに変換して、golangから参照するところまで紹介しました。次回はtreeliteのモデルとLightGBMのモデルとの予測速度比較、golangから参照したtreeliteモデルとgolangで記述されたアンサンブルモデル予測ライブラリleavesとの予測速度比較を行いたいと思います。

treeliteのドキュメント

SQS + Lambdaでs2sの基盤を作った話

こんにちは。技術開発部・配信/インフラチームの二階堂です。

弊社のプロダクトでは連携している効果測定ツールに対して、S2Sで通知を行う場合があります。この通知処理を以前まではEC2上に配置したデーモンプログラムで行っていたのですが、処理の共通化・高速化・インフラ費用削減などの目的でSQS+Lambdaに移行しました。この経験を踏まえて移行時のポイントを紹介したいと思います。

移行前のフローと問題点

通知先のURLの入ったログを通知用インスタンス(複数ある内のどれか一つ)に書き出す
EC2上に配置したデーモンプログラムがログを読み込み通知を行う
同じくデーモンプログラムが通知結果のログを書き出す

問題点

常時起動のインスタンスが高価
時間帯によって通知量が変化するため通知量が多い時に時々遅延が発生する
通知漏れが発生した際にどの通知用インスタンスが原因か調査するコストが高い

SQS + Lambda のフロー

通知先のURLの入ったログをSQSに送信
定期実行されるLambda(図中a)がSQSに送られたメッセージ数(NumberOfMessageSent)に比例した数の通知用Lambda(図中b)をキック
通知用LambdaがSQSからメッセージを取得→通知→結果をS3に保存を5分間繰り返す

特徴・改善点

常時起動EC2よりLambdaの方が圧倒的に安価 (問題点1)
全てのログを一度SQSに送る事でそれ以降のフローを共通化し調査コストを下げた (問題点3)
Lambdaを2段構成にする事で通知量の変化に柔軟に対応できるようになった (問題点2)
通知用Lambdaはログのパースと通知しかしていないので他の通知にも容易に対応できる

効果

インフラコストが約10分の1に減った
リリース以前には定期的に発生していた調査や再通知などの対応もほぼ無くなり安定して動作している

移行した感想

Lambdaの料金は実行時間*メモリ使用量なのでその辺を意識したコードになっていないと高い効果が得られません。今回の処理内容はただの通知なのでメモリはあまり使わないので速度を上げる工夫として通知部分を並列化しました。最初は並列化していなかったのですがその時は移行前とあまり費用が変わらなかったことを考えるとコスト意識の重要性が判りやすいのではないかと思います。

今回の基盤開発は並列化対応も含めて何かと初めての経験が多い開発だったので効果にしても経験にしてもとても有意義なものだったと思います。

今回は以上となります。

Lambda@Edgeを利用した画像のリサイズを試してみた

こんにちは。技術開発部部長の川住です。

弊社プロダクトにて、「オリジナル画像を適切な大きさに縮小してから読み込む仕組み」を作る必要が生じたので、遅ればせながらLambda@Edgeを使ってみたので、利用時のポイントを紹介したいと思います。

Lambda@Edgeとは？

Lambda@Edgeは、Amazon CloudFrontのエッジサーバで、AWS Lambdaの関数を実行できる仕組みです。リクエスト-レスポンス間の以下4つのタイミングに対してLambda関数を設定できます。

CloudFrontのエッジサーバがユーザからのリクエストを受け取る時 (Viewer Request)
(エッジサーバにキャッシュがなく)、オリジンサーバへリクエストを転送する前 (Origin Request)
オリジンサーバからレスポンスを受信した後 (Origin Response)
エッジサーバからユーザへレスポンスを転送する前 (Viewer Response)

Lambda関数に機能を実装することで、例えば以下のような仕組みを実現できます。

ユーザ認証やリダイレクトを行う
HTTPヘッダを操作する (追加, 削除, 変更)
コンテンツを必要に応じて加工してレスポンスする

Lambda@Edgeを使用した画像のリサイズ

Lambda@Edgeを利用した画像のリサイズの仕組みの構築方法に関しては、こちらにて詳しく紹介されていますので、実装の詳細については割愛します。大雑把に言いますと、2つのLambda関数を実装しそれらをCloudFrontに設定することで実現しています。

Viewer Requestとして、リサイズ後の幅や高さなどのパラメータを受け取り、パスに組み込む関数をCloudFrontに登録する
- エッジにリサイズ後の画像のキャッシュがあれば、オリジンへリクエストは転送されず、キャッシュデータがレスポンスされる
Origin Responseとして、必要に応じてオリジンから受け取った画像をリサイズし、レスポンスとして返す関数をCloudFrontに登録する
- オリジンにリサイズ後の画像がすでに保管されている場合は、保管されている画像をレスポンスする
- オリジンにリサイズ後の画像が保管されていない場合は、画像を取得してリサイズした後、オリジンにデータを格納し、画像をレスポンスする

Lambda@Edge利用時のポイント

Lambda@EdgeはAWS Lambdaと同様の制約を受けるほかに、Lambda@Edge特有の制約や注意点があったので紹介します。

スペック面での制約

AWS Lambdaでは実行時のメモリ量やタイムアウト時間を設定できますが、Lambda@Edgeでの実行時にはこれらの設定上限が通常よりも厳しくなります。

メモリ量の上限: 128MB
タイムアウトの上限: 5秒

その他制約に関してはこちらが詳しいです。また、AWS Lambdaでは複数の言語をサポートしていますが、Lambda@Edgeで使用できるのは「 node.js 6.10, node.js 8.10 」のみとなっています。

Lambda関数はus-east-1 (バージニア北部) に登録する必要がある

CloudFrontに設定できるLambda@Edge関数はバージニア北部に登録されているものだけです。

バージョンつきのarnを紐つける必要がある

CloudFrontには $LATEST や $ALIAS のarnを設定できないので、バージョンを発行したLambda関数をCloudFrontに設定する必要があります。

実行時のCloudWatch Logはエッジサーバのリージョンに排出される

Lambda関数の登録自体はus-east-1 (バージニア北部) ですが、実行はCloudFrontのエッジサーバであるため、実行時のCloudWatch Logの排出先はエッジサーバのロケーションに依存します。 (日本からアクセスした場合には、東京、ソウル、シンガポールリージョンあたりにログが排出されていました。大半は東京。)

使ってみた感想

今回の用途以外にも、A/Bテストなどにも利用できそう
大量アクセスのある場面には注意が必要
- リクエスト数とLambdaのコード実行時間に比例して利用料金が発生。キャッシュヒット率を高める工夫が必要になってきそう。

今回は以上となります。

AWS ストレージサービス「S3」

今週もまた台風が近づいておりますね。初めまして、今年新卒で入社した技術開発部の程です。週末は同僚と登山に行く予定なのに台風直撃の知らせを聞いて下がり気分です。自然ばかりはITの技術だけではどうにもならないので仕方ないですね。

さて、私、現在は自社DSP、Bypassの開発に携わっているのですが弊社広告配信システムはAWSを利用して構築されております。今回は、AWSの中でも特に弊社広告配信システムにおいても実際に活用している S3というストレージサービスについて簡単に概要や特徴などについてまとめたいと思います。

S3

S3では、バケットと呼ばれる仮想的なオブジェクト置き場に様々なファイルやメディアコンテンツ（＝オブジェクト）を格納し AWSクラウドサーバ上に保存することができます。保存したオブジェクトはいつでもどこからでも参照したり新規オブジェクトを追加することができます。

S3の特徴

容量無制限
スケーラブル
容量は無制限となっており、大量のデータを扱う場合にも、上限を心配することなく使用できます。
冗長化されている
保存したオブジェクトは異なるアベイラビリティゾーン（AWSによる場所の区分け）にある複数のサーバーに複製保存されるので障害発生時にデータロストの危険性が低く安心です。また、自前で冗長化する手間が省けるという利点もあります。
高い堅牢性
99.999999999%（イレブンナイン）の堅牢性（aws公式サイトより）

S3の操作

S3上でオブジェクトは次のような形式で表現され、管理されます。

s3://{バケット名}/{オブジェクトを保管しておく場所を表すパス}

1	s3://{バケット名}/{オブジェクトを保管しておく場所を表すパス}

調べたところ、S3内部ではフォルダという概念はなく、S3上では単純にKey-Value方式でオブジェクトが格納されているのだそうです。とはいえ、実際に使ってる時はあんまり意識する必要はなく、通常のファイルと同じような感覚で操作できるのがいいですね。

AWSではS3を操作するための方法をいくつか用意しており、目的に応じて使い分けも可能です。具体的には以下のようなものが挙げられます。

ブラウザ上のコンソールから操作
コマンドラインから操作
AWS提供のSDKより各種アプリケーションから操作
3rd party製のツールで操作

コマンドラインから操作できるのは便利ですね。以下のように、基本的なバケットやファイルの作成、削除、コピーなどの操作は一通りCLIで実行可能です。

$ aws s3 mb s3://{バケット名}                          #バケットの作成(おそらくmb=make bucket)
$ aws s3 rb s3://{バケット名}                          #バケットの削除(おそらくrb=remove bucket)
$ aws s3 sync {フォルダパス} s3://{バケット名}/{パス}　     #バケットの内容をローカルフォルダと同期
$ aws s3 cp {ファイルパス} s3://{バケット名}/{パス}        #ローカルのファイルをバケットにコピー
$ aws s3 rm s3://{バケット名}/{ファイルパス}              #バケットのファイルを削除

$ aws s3 mb s3://{バケット名} #バケットの作成(おそらくmb=make bucket)

$ aws s3 rb s3://{バケット名} #バケットの削除(おそらくrb=remove bucket)

$ aws s3 sync {フォルダパス} s3://{バケット名}/{パス}　 #バケットの内容をローカルフォルダと同期

$ aws s3 cp {ファイルパス} s3://{バケット名}/{パス} #ローカルのファイルをバケットにコピー

$ aws s3 rm s3://{バケット名}/{ファイルパス} #バケットのファイルを削除

S3操作用のコマンドとしては上記のものの他に、aws s3apiというものがあり、前者はAPIとコマンドが１：１で対応する形の低レベルなコマンド群であるのに対し、後者は複数のリクエストにまたがる様な処理などが実装された高レベルなコマンド群である、という違いがあります。

もちろん、ローカル->S3だけではなく, S3->ローカルやS3->S3間のファイル移動を行うこともできます。

コンソールから操作する場合は以下のような画面上で、バケット作成やファイルのアップロードを行います。

ログインしたら最初に表示されるバケット管理画面のイメージです。

バケット新規作成の後は、そのバケット内に格納されるファイルを管理する画面に遷移します。この画面上では、バケットへのファイルのアップロードや、フォルダの作成などができます。

適当に１個ファイルをアップロードしてみるとこのようにファイルが追加されます。

コンソール上で操作する場合は、クリック操作だけで非常に簡単に、バケットの作成からファイルのアップロードまで実行することができました。

S3のユースケース

AWS公式サイトやWebサイト上の記事を調べると、次の３つが主なユースケースとしてよく取り上げられています。

データのバックアップ
前述の通り、高い堅牢性を備えていることから、消失すると困る様々なデータを保管する。
コンテンツ配信
S3上に保存したコンテンツを配信する。
ログデータなどの保存先
EC２で収集されたログの退避先、ビッグデータ分析で使用する生データの保存先として利用する。

弊社広告配信システムにおいては特に動画広告の場合に、S3上に動画素材をアップロードした上で AWS CloudFrontとS3を連携させることで、容量の大きい動画系広告でも高速かつ安定した広告配信を実現しています。

バージョン管理

S3で保存されるオブジェクトは、バージョンで管理することも可能です。例えば、誤削除などのミス発生時などにこの機能が適用されていれば、すぐに以前のバージョンに復旧することができます。この機能を有効化すると、オブジェクトを更新した時などに前の世代のオブジェクトが自動的に保管される様になります。何世代分保存するかを指定することもできます。

Notification(通知)機能

バケットにファイルが追加されたことをイベントとして検知したい時はこの機能が有用です。この機能ではバケット単位で、以下のイベントが発生した際に Amazon SNS, SQS, LambdaといったAWSサービスに通知を飛ばすことができます。

新しいオブジェクトの作成イベント(Put, Post, Copy, CompleteMultiPartUpload)
オブジェクト削除イベント(Delete, DeleteMarkerCreated)
低冗長化ストレージのオブジェクト消失イベント(RRSObjectLost)

その他の機能

クロスリージョンレプリケーション
別リージョンへの複製保存を行う。
S3へのアクセスログ
S3上でバケットに対してどんな操作を行ったかの記録ログを出力させる。
Tag管理
バケットに対してタグを指定する。
メタデータ
オブジェクトに対してメタデータを設定する。

私自身、入社して初めてクラウドサービスを触ったのですが実際に使ってみるとその操作の簡単さと便利さに驚きました。個人利用も可能なので、個人の自主アプリ制作などにも活用できそうですね。

それでは以上となります。

AWS Glue + Athena構成を試す

こんにちは。技術開発部の赤井橋です。

弊社では現在adstirログ基盤のリプレイスを計画しており、その一貫としてAWS Glueでのデータ変換（json → parquet）、及び変換データのAthenaでの検索を試しました。

Glueとは

https://aws.amazon.com/jp/glue/

2017-08-15から利用出来るようになった抽出、変換、ロード (ETL) を行う完全マネージド型のAWSサービスです。使い所としてはファイルのカラムナフォーマットへの変換、及びパーティションが有効なディレクトリへの配置、データカタログ(テーブル定義のメタデータ)の更新・・など、ビッグデータを使いやすく成型する場面が多いかと思います。

Glueが提供する機能

大きく分けて2つ存在します。(2018-07-31時点)

1. データカタログの更新

クローラという機能を用いてデータカタログの自動更新を行うことが出来ます。クローラは指定されたデータソース（特定のS3ディレクトリなど）内をスキャンし、該当ディレクトリにあるファイルフォーマットに合わせたテーブル定義（パーティショニング含む）を自動で行ってくれます。定義されたデータカタログはAthena、EMR、Redshift Spectrumでも使用(2018-07-31時点)でき、実行スケジュールの登録も可能です。

2. ETL

ジョブという機能を用いてデータ抽出、変換を行うことが出来ます。ジョブ追加時のセットアップガイダンスに従って進めていくと最終的にPythonのスクリプトが自動生成されます。シンプルな処理であればスクリプトの修正は不要ですが、手の込んだ処理の場合修正する必要があります。ジョブを定期実行するトリガーという機能もあります。

使用にあたり

上記2つの機能はどちらかだけ使用してもよく、どちらとも使用したい場合も使用順序に制約はありません。そのため、クローラでテーブル定義を更新し、更新されたテーブル定義を元にジョブでのデータ変換を行う、という流れ以外にもジョブで変換されたファイルに対してクローラでテーブル定義を更新する、という用途でも使用出来ます。

ジョブのスクリプト

セットアップガイダンスを終えるとPythonスクリプトが生成されますが、例えば特定の引数を受け取って処理を行いたい場合は別途実装が必要です。スクリプトではGlueで独自定義されたDynamicFrameというデータ構造を操作することで独自の変換処理を行えます。

例）AWS CLIからyear引数を指定してGlueを実行し、受け取ったyear引数をS3のデータソースのパスとして設定したい場合

AWS CLI

aws glue start-job-run --job-name TestJob --arguments '{"--year":"2018"}'

1	aws glue start-job-run --job-name TestJob --arguments '{"--year":"2018"}'

スクリプト

from awsglue.utils import getResolvedOptions

# 引数からパスを生成
args = getResolvedOptions(sys.argv, ['year'])
basepath = "s3://mybucket/Glue/year="
s3path = basepath + args['year']

sc = SparkContext()
glueContext = GlueContext(sc)

# 作成したパスのデータを指定して、DynamicFrameを作成
datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options={"paths": [s3path]}, format="csv", format_options={}, transformation_ctx="datasource0")

...
#以後、DataFrameに対する処理を記述

from awsglue.utils import getResolvedOptions

# 引数からパスを生成

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

basepath = "s3://mybucket/Glue/year="

s3path = basepath + args['year']

sc = SparkContext()

glueContext = GlueContext(sc)

# 作成したパスのデータを指定して、DynamicFrameを作成

datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options={"paths": [s3path]}, format="csv", format_options={}, transformation_ctx="datasource0")

...

#以後、DataFrameに対する処理を記述

変換対象のフォーマットが複雑な場合も別途実装が必要です。

例）tsvフォーマットの1列にjson文字列があり、json文字列の部分をstructの配列として変換したい場合

元データ（tsv）

"2018-07-31 00:00:00"   "type1" "{column3:[{sub_column1: xxxx, sub_column2: yyyy}, {sub_column1: xxxx, sub_column2: yyyy}]}"
"2018-07-31 00:00:00"   "type2" "{column3:[{sub_column1: xxxx, sub_column2: yyyy}, {sub_column1: xxxx, sub_column2: yyyy}]}"

1 2	"2018-07-31 00:00:00" "type1" "{column3:[{sub_column1: xxxx, sub_column2: yyyy}, {sub_column1: xxxx, sub_column2: yyyy}]}" "2018-07-31 00:00:00" "type2" "{column3:[{sub_column1: xxxx, sub_column2: yyyy}, {sub_column1: xxxx, sub_column2: yyyy}]}"

スクリプト

# マッピング
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("col0", "string", "col0", "string"), ("col1", "string", "col1", "string"), ("col2", "string", "json", "string")], transformation_ctx = "applymapping1")

default_df = applymapping1.toDF()

# スキーマ定義
fields = [
    StructField("column1",StringType()),
    StructField("column2",StringType()),
    StructField("column3",ArrayType(
        StructType([
            StructField("sub_column1", StringType()),
            StructField("sub_column2", IntegerType()),
        ]))),
]
schema = StructType(fields)

# データ成型
def format(col0, col1, json):
    json_data = json.loads(json)
    ret = {}
    ret["column1"] = col0
    ret["column2"] = col1
    json_list = []
    for sub_column in json_data["column3"]:
        sample_dict = {}
        sample_dict["sub_column1"] = str(sub_column["sub_column1"])
        sample_dict["sub_column2"] = int(sub_column["sub_column2"])
        # append
        json_list.append(sample_dict)
    ret["column3"] = json_list
    return ret

# データ成型してDynamicFrameにする
rdd = default_df.select("col0", "col1", "json").rdd.map(lambda x: format(x.col0, x.col1, x.json))
sqlContext = SQLContext(sc)
parsedDF = sqlContext.createDataFrame(rdd, schema)
dyf = DynamicFrame.fromDF(parsedDF, glueContext, "sample_format")

# マッピング

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("col0", "string", "col0", "string"), ("col1", "string", "col1", "string"), ("col2", "string", "json", "string")], transformation_ctx = "applymapping1")

default_df = applymapping1.toDF()

# スキーマ定義

fields = [

StructField("column1",StringType()),

StructField("column2",StringType()),

StructField("column3",ArrayType(

StructType([

StructField("sub_column1", StringType()),

StructField("sub_column2", IntegerType()),

]))),

]

schema = StructType(fields)

# データ成型

def format(col0, col1, json):

json_data = json.loads(json)

ret = {}

ret["column1"] = col0

ret["column2"] = col1

json_list = []

for sub_column in json_data["column3"]:

sample_dict = {}

sample_dict["sub_column1"] = str(sub_column["sub_column1"])

sample_dict["sub_column2"] = int(sub_column["sub_column2"])

# append

json_list.append(sample_dict)

ret["column3"] = json_list

return ret

# データ成型してDynamicFrameにする

rdd = default_df.select("col0", "col1", "json").rdd.map(lambda x: format(x.col0, x.col1, x.json))

sqlContext = SQLContext(sc)

parsedDF = sqlContext.createDataFrame(rdd, schema)

dyf = DynamicFrame.fromDF(parsedDF, glueContext, "sample_format")

GlueとAthena、使ってみて不自由だった点

Glueでデータソースとして読み込めるのは文字コードがUTF-8のみ

UTF-8以外の文字列が変換対象のデータソースに含まれているとGlueでの変換処理が失敗します。

[AWS Black Belt Onine Seminar] AWS Glue

根本的な解決ではないですが、AWS Lambdaを用い対象となるログファイルをUTF-8に変換する、という前処理を行うことで対処しました。

DPUを上げても劇的な処理速度向上は見込めない

GlueではDPU(データ処理ユニット)の数を指定出来るのですが、2倍にすれば2倍処理が早くなる、という挙動ではありませんでした。 7.7G分のファイルでそれぞれ検証したところ、

DPU 10 → 14 mins
DPU 50 → 9 mins
DPU 100 → 9 mins

DPU 10 → 14 mins

DPU 50 → 9 mins

DPU 100 → 9 mins

という結果でした。

Athenaでのstruct型の使い勝手が悪い

struct型のスキーマ定義に値を追加すると、スキーマと同様の構造で格納されているパーティションでは問題なく検索出来ますが、値が存在しない(スキーマ変更前の構造の)ファイルが格納されているパーティションでは「HIVE_CANNOT_OPEN_SPLIT」エラーが発生し検索できませんでした。

この定義から 
 CREATE EXTERNAL TABLE `test_tbl`(
  `id` int, 
  `column` array&lt;struct&lt;aaa:int, bbb:int, ccc:int>>,
 〜〜

この定義に変更すると、定義変更前のファイルを読み込もうとした際にHIVE_CANNOT_OPEN_SPLITエラーが発生
CREATE EXTERNAL TABLE `test_tbl`(
  `id` int, 
  `column` array&lt;struct&lt;aaa:int, bbb:int, ccc:int, new_column:int>>,
 〜〜

この定義から

CREATE EXTERNAL TABLE `test_tbl`(

`id` int,

`column` array<struct<aaa:int, bbb:int, ccc:int>>,

〜〜

この定義に変更すると、定義変更前のファイルを読み込もうとした際にHIVE_CANNOT_OPEN_SPLITエラーが発生

CREATE EXTERNAL TABLE `test_tbl`(

`id` int,

`column` array<struct<aaa:int, bbb:int, ccc:int, new_column:int>>,

〜〜

こちら根本的な解決法はないようで、暫定対応としてstruct型をstring型に変更し検索の際にstring文字列をjsonと扱うように対処しました。

まとめ

検証の結果、コストや処理時間なども考慮するとまだGlue + Athenaを導入する判断に至っていません。ただ、比較的新しく提供されたサービスのため今後も機能が追加され使い勝手も改善されていくはずです。ビッグデータがますます全盛となる時代、このようなサービスの重要性は増していくのではないでしょうか。

今回は以上です。