Shoryukenでつくるバッチ処理基盤

トレタのAPI開発を担当している芹沢です。

トレタでは、長時間かかるバッチ処理を複数台のサーバ上で処理させて短時間で処理できるバッチ処理基盤をAWS上で構築しました。この仕組みについて説明します。

目的

短期的には以下の課題を解決するため、長期的には似たような要件が再度発生した時に、同じ手法で解決できることを目的に作りました。

• 非同期でDBをデータソースとしたデータを加工してCSVファイルとして出力してS3にputしたい
  • データソースはDBに入っているリアルタイムのデータであることが求められる
  • CSVファイルの作成は決められた時間内に完了する必要がある
  • 対象となるデータソースの量は日々増加し続けるが、常に決められた時間内にCSV作成が完了している必要がある

難点

今回の要件で技術的に難しい点は以下の２点です。

DBを直接参照しながら大量のデータを処理する

例えば、データソースとしてDBからHDFSやRedshiftに同期されたデータを使えるのならば、AWS EMRなどのサービスを活用することで大量のデータを処理することは比較的容易です。しかし、今回の要件は、更新頻度が高いデータソースをできるだけリアルタイムに近い状態で取得してCSVを作成する必要がありました。

一定のスループットを担保し続ける

今回の要件上、1回のCSV処理作成にかかる時間を15分以内に抑える必要がありました。inputとなるデータ量が常に一定であれば一度15分以内に終わるように構築すれば、あとはそれを延々と動かし続ければ良いのですが、データ量が増加し続けるという事情があるため、データ量が増えても一定のスループットを担保し続ける仕組みを用意する必要がありました。

バッチ処理専用の環境を作ることにしました

これらの課題を解決するために以下の様な基盤を構築し、その上にデータ処理用のバッチを実装しました。

処理の流れは図の通りです。

バッチの実装について

いわゆるJob Queueですが、以下２つのJobが介在します。前者はSidekiq、後者はShoryukenを使って実装されています。

IDをグルーピングしてCSV作成Jobを作るJob

1. sidekiq-cronによって15分に1回起動します
2. 処理対象のレコードのIDを取得し、100IDずつにグルーピングします
3. グルーピングしたIDを引数としたCSV作成Jobを処理対象のレコード数分だけ作成します

CSVを作成するJob

1. Jobに含まれるIDを元にCSVを作成します
2. 作成したCSVをS3にpushします

1st release時点では、処理対象のデータのIDが800個ほどあったので、１回の処理につき8個のCSV作成Jobが生成され、8個のShoryuken workerが8並列で処理します。

個々のCSV作成Jobの処理時間は概ね10分〜11分に収まっているので、結果的に15分以内に処理を完了させるという目標を達成できました。処理対象のIDが増加してもShoryuken orkerの数を増やすことでスループットを増加させることが可能になっており、スケールアウトしやすい構造になっています。

Shoryukenについて

あまり日本語の記事を見かけなかったので簡単に紹介します。

Sidekiqとほぼ同じような使い方で使えるworker gemです。Sidekiqと異なる点として、Queue storeにAWS SQSを使います。元々トレタでは非同期処理にはSidekiq + redisを使っていましたが、今回のようにJobの数がスケールしていくことが想定されるケースにおいてはQueueのサイズや数に融通が効くSQSの方が適しているという判断の元、SQSを使うためにShoryukenを採用しました。

Tips: 各Jobのロギングについて

ShoryukenにはJobの実行時に前処理・後処理を挟むなどの目的に使えるMiddleware機構があります。 今回はJob実行時のパフォーマンス計測とエラー時の原因調査のために各Jobのログをfluentdで拾ってBigQueryに入れておきたかったので、以下のようなMiddlewareを自作してログをテキストファイルで残すようにしました。

module Shoryuken
  module Middleware
    module CustomMiddleware
      module Server
        class ProcessingTimeMeasure
          def call(worker, queue, sqs_msg, body)
            shoryuken_log_path = Rails.root.join('log', 'shoryuken_worker.log')
            @logger = ::Logger.new(shoryuken_log_path)
            @logger.formatter = proc do |_, _, _, message|
              message.to_json + "\n"
            end

            @success = true
            @messages = []

            started_at = Time.zone.now

            # ここでJob本体の処理が実行されます
            yield

            ended_at = Time.zone.now

            elapsed_time = (ended_at - started_at) * 1000
          rescue => e
            @success = false
            @messages << e.to_s
            Bugsnag.auto_notify(e)
          ensure
            body = {
              timestamp: Time.zone.now.to_i,
              started_at: started_at,
              ended_at: ended_at,
              elapsed_time: elapsed_time,
              success: @success,
              messages: @messages,
              worker: worker.class.to_s,
            }
            @logger.info(body)
          end
        end
      end
    end
  end
end

このMiddlewareをShoryukenに積むにはconfig/initializes配下に適当なファイルを置いて以下のように追記します。

Shoryuken.configure_server do |config|
  config.server_middleware do |chain|
    chain.add Shoryuken::Middleware::CustomMiddleware::Server::ProcessingTimeMeasure
  end
end

まとめ

今のところ安定して稼働していますが、基盤としてはまだ汎用的なつくりにはなっていないので、汎用基盤として稼働させていくためには以下のような機能が必要になりそうだと考えています。

• 処理時間に制限があるJobの時間が経過した場合に通知する(Shoryuken workerの数を増やすタイミングを見極めるための何か）
• Job管理(workflow系ツールの導入など？）
• Shoryukenのプロセス監視と自動復旧

今後も開発を続けながら、何かシェアできる知見が得られたらまた開発者ブログに書きたいと思います。

お約束

トレタはエンジニアを募集しています。iPadアプリやAPIの開発以外に、こういった基盤開発にも興味があるエンジニアはぜひご応募ください。

www.wantedly.com

www.wantedly.com

テストデータ生成に欠かせない便利な◯◯kitの勧め

iOSを担当している高です。

開発しているとそれっぽいテストデータが欲しいってことが結構あります。今までは各々がローカルで都度スクリプト書いてるような状態で、僕の場合はトレタアプリの中に直接書いて都度書き捨ててる感じでした。 これは相当効率が悪いですし、自分用に書いたものは人にも共有しづらいという問題があったのでこれを機にツールを作ってみることにしました。

開発上の課題

iOS開発をする上でこの様な課題がありました。

• パフォーマンス確認のために大量データが欲しい
• テーブル数などの設定をかんたんに変えたい
• 外部連携が必要なテストデータが欲しい。けれどE2Eで作るのはちょっと大変

大量データは文字のごとくです。通信時間やアプリ上での描画パフォーマンスを確認するために想定される最大のデータを用意したい。データ作成が1回だけで良ければ手で作ってもいいのですが、トレタの場合日付が軸になるので次の日には過去データになってしまいます。そうならないように未来データとして作ると今度は毎回そこまで行くのが面倒という本末転倒なことになります。

テーブル数も現状アプリからは1テーブルずつでしか追加と削除が出来ません。これはユーザが使う分には問題無いのですが、テストで多様なデータを用意する場合には結構手間な作業になります。消すのも同様です。

外部連携が必要なテストデータに関してはアプリ上ではまずは表示の確認がしたいだけのことが多いです。charlesでレスポンスを書き換えたりしていたのですが、結構手間なこともあり、あまり実用的ではありませんでした。

octokit

言わずと知れたGitHubのAPIツールです。 https://octokit.github.io

コマンドラインからちょっとしたことをやりたい時なんかにoctokit.rbを使ったりしますよね。こんな感じでトレタのAPIも触れるといいなぁと思いました。

toretakit

ということでtoretakit inspired by octokitを作ってみました。

octokit.rbと同じようにgemとして作っています。

予約データを作る

トレタで予約を作るために必要な最低限の情報はこれらになります。

• 予約日時
• 氏名（漢字）
• 氏名（よみがな）
• 電話番号
• 人数

氏名や電話番号にはfakerというgemを使いました。結構有名なgemなのでご存じの方も多いと思います。ただ、氏名に関してはfakerはふりがなに対応していないので、gimeiというgemを使いました。 gimeiの説明をREADMEから引用します。

gimei は、日本人の名前や、日本の住所をランダムに返すライブラリです。テストの時などに使います。似たようなライブラリにfakerがあります。fakerはとても優れたライブラリで、多言語対応もしていますが、ふりがな(フリガナ)は流石に対応していません。gimei はふりがな(及びフリガナ)に対応しています。

出現率の異なるランダム値が欲しい

テストデータを作るときに同じデータばかり作ってもしかたがないのである程度ランダムで作りたいことがあります。

例えばトレタの場合だと予約するときの人数はランダムで決めたい、みたいな感じです。ただこの時に本当に単純なランダムにしてしまうと2人の予約と10人の予約が同じ回数だけ出てくるという現実離れしたデータになってしまいます。

実際のデータではところどころ山があると思いますが、そこまで忠実に再現するのは既存データを洗ったりしないといけなくてちょっと大変なので簡単に大きい数字の方が出にくいようにしたいと思いました。

なんか良いアルゴリズムが無いかなと探したのですが見つからず、とりいそぎベタにこんな感じで対応してみました。数字の間隔はなんとなくです。

def rand_count()
  r = rand(100)
  case r
  when 99..100 then 10
  when 96..99 then 9
  when 93..96 then 8
  when 90..93 then 7
  when 85..90 then 6
  when 80..85 then 5
  when 70..80 then 4
  when 60..70 then 3
  when 50..60 then 2
  when 0..50 then 1
  end
end

何か良いアルゴリズムがあったら是非教えてほしいです！

どんなことが出来るようになったか

こんな感じでテーブル数が100の設定で、2回転しているデータ（かなりの繁盛店！）をそれっぽく作る、みたいなことが簡単にできるようになりました。

（人数が固定になってますが、先ほどのランダムロジックはここではまだ使ってなくて他のところで使ってます）

また、他のPJでも使ってもらえるようになりました。

今後の展望

現状toretakitはAPI通信とデータ生成が一緒くたになっているのでのちのち別gemとして切り出したいなと思ってます。

将来的には複数店舗さんのシミュレーションデータ生成ロジックを用意して、コマンド一発でそれっぽいデータが用意できるようにしたいなと思っています。

まとめ

僕は普段はSwift、たまにObjective-C（既存コード）という生活を送ってるのですが、こういったツール系で普段と違う言語・環境で開発するのも楽しいものですね。

引き続きエンジニアは募集中ですので興味のある方はチェックしてみてください！

メールの配信状況を可視化、追跡する

週1でスープカレー食ってる佐野です。仕事ではトレタのインフラをあれこれしています。今回はメール配信の異変にいち早く気づき、カスタマーサポートのレスポンスを向上する取り組みについてです。 スマートフォンの普及、メッセンジャーの台頭などによって個人間でのメールでのやりとりは減っているかもしれませんが、通知の仕組みとしてまだまだメールは現役です。弊社ではお店への予約確定の通知、お店への予約一覧のPDF送信、お客様への来店日のリマインド...などにメールを活用しています。メールを使っていると、たまにお客様から弊社カスタマーサポートに「メールが届かない」「突然届かなくなった」という問い合わせをいただくことがあります。担当者は原因(トレタの障害？メール配信システムの障害？お客様のメールアドレス間違い？...etc)を即座に調べて回答する必要があります。今日はその仕組みについて。技術的には簡単な話です。

メール配信

トレタではSendGridを使っています。SendGrid自体の障害に備えて、コンフィグの変更/デプロイでAWS SESに切り替えられるようにもしてあります。SendGridには処理したメールの状況を通知できるEvent Notificationという仕組みが備わっていて、各種イベント(Processed(メール送信リクエスト受け付け)、Delivered(受信側メールサーバまで到達)、Deferred(遅延)、Bounces(拒否)、Drops...etc)とその理由を通知できるので、これを利用してメールの状況を取得/可視化します(SendGridのコンソールでも配信状況は見れるのですが直近数日分しか見ることができません)。ちなみに、SESでもSNSに通知が可能なので、SESでも似たような仕組みは簡単に作れるはずです。

仕組み

その図です。

AWS上で稼働するFluentdでEvent Notificationを受け付け、それをBigQueryに投入することでSQLで検索できるようにしてあります。またMackerelでグラフを描いて可視化しています。

https+basic認証でイベントを通知する(Event Notification)

こちらはSendGridのコンソールから簡単にできます。宛先のURLを入力するだけです。

SendGridからのEventの受信

拙著fluent-plugin-sendgrid-eventを利用しています。SendGridから飛んできたHTTP POSTをパースします。すでにプラグインがありそうだったけどなかったので書いてみました。SendGridのEvent NotificationはHTTP 503を返せばしばらく時間をおいてから同じイベントを再送してくれるので、このプラグインでは予期せぬエラーが発生した際は503を返すようにしてあります。

BigQueryへの投入

fluent-plugin-anonymizerとfluent-plugin-bigqueryを利用しています。データにはお客様のメールアドレスが含まれているためそれをfluent-plugin-anonymizerでハッシュしたうえで、BigQueryに転送します。お客様から問い合わせのあったら、そのメールアドレスをsaltとともにハッシュ関数に通してSQLを叩くことで確認できます(管理画面作らんと...)。

※ メアドはハッシュしているので問題はないはずですが、画像では一応マスクもしてあります。

Mackerelでの可視化

fluent-plugin-datacounterとfluent-plugin-mackerelを利用しています。次の画像は、ちょうど先日、SendGridで発生したメール配信遅延障害時の状況です。ある時間帯にdeffered(遅延:紫のグラフ)が大量に発生していることが確認できます。status.sendgrid.comに情報が出るよりも早く、遅延が発生していることに気づくことができました。

簡単ですが以上です。

おわり