この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
ウィスキー、シガー、パイプをこよなく愛する（最近ビールへ浮気が多い）大栗です。
2016年6月18日に開催された第五回ゲームサーバ勉強会に参加してきたので、レポートしたいと思います。個人的にはDigdagに興味があり、githubへの公開後初のセッションの様だったので参加してみました。
EmbulkやDigdagのお話
Treasure Data Support Engineering ManagerのToru Takahashi（@nora96o）さん
はじめに 〜 データ分析基盤について 〜
オンラインゲームシステムとデータ分析基盤の関わり
データ分析基盤をなぜ必要なのか？
- 現状を知る
- データ分析の流れをスムーズに行う
- データの民主化
データ分析処理エンジンと様々なマーケティング施策サービスとの連携が大きな役割。
データ分析基盤を支えるOSS
- Fluentd：データストリーミングコレクタ
- Embulk：バルクデータローダ
- Digdag：ワークフローオートメーションシステム
お知らせ）Fluentd、ロゴかわるってよ
TreasureData：データ分析基盤を一気通貫で提供
Embulk 〜 Bulk Data Loader 〜
Embulkとは？
- オープンソースのバルク転送ツール
- プラグイン機構
- データ連携を容易に
バルクデータ転送の難しさ
- 入力データの正規化
→ 試行錯誤してデータを正規化する - エラー処理
→ データバリデーション、リトライ、レジューム - メンテナンス
→ ドキュメント、汎用化、OSS化 - パフォーマンス
→ 並列・分散処理
バルクデータ転送の例
指定された 10GB CSV file をPostgreSQLにロード
- コマンド叩いてみる → 失敗
- データを正規化するスクリプトを作成
”20150127T190500Z” → “2015-01-27 19:05:00 UTC”に “null” → “N”に変換
元データを見ながら気付く限り… - 再度チャレンジ → 取り込まれたが元データと合わない
“Inf” → “Infinity”に変換 - ひたすら繰り返す
- うっかりレコードが重複して取り込まれた...
- スクリプトが完成
- cronに登録して毎日バルクデータロードするよう登録
- ある日、別の原因でエラーに…
不正なUTF-8 byte sequenceをU+FFFDに変換
過去の日次 10GB CSV file を 730個 を取り込む（2年分）
- たいていのスクリプトは遅い
最適化している暇がない 1ファイル1時間、エラーが発生しなくても1ヶ月必要 - 並列データロードするようにスクリプト変更
- ある日、ディスクフル / ネットワークエラーで失敗 どこまで読み込まれた？
- 障害後に再開し易いよう、データロード単位を調整
- 安全な再開機能をスクリプトに追加
システム構築の頭痛の種
様々な転送データ、データストレージ
N x M のスクリプトが必要！
Embulkのアプローチ
- プラグインアーキテクチャ
- 入力データ正規化支援: guess, preview
- 並列・分散実行
- 繰り返し実行
- トランザクション制御
プラグインアーキテクチャ
- フレームワークによる恩恵
- 並列処理、繰り返し実行、エラー処理、リカバリ
- RubyGemとして配付
- DB
- SaaS
- File Format
DEMO - Embulk
CSVファイルをPostgreSQLへロード
（ネットワーク環境が悪いとよくトラブるので、ネットワークは基本使わないようにしているそうです）
Embulkプラグインアーキテクチャ概要
4種のプラグインとそれを組み上げるフレームワーク
- Executor: 実行
- Input: バルクデータのレコード群を取り込み
• FileInput, Decoder, Parser: ファイル操作 - Filter: レコードに対するデータ操作
- Output: バルクデータのレコード群を出力
• FileOutput, Encoder, Formatter: ファイル操作
TreasureDataでのEmbulkのユースケース
Embulkで解決できないこと
YAML管理- 単体処理フレームワーク • 事前処理・事後処理
- ジョブキュー
並列数制御- 続きから実行 • 処理間の依存関係
参考: Embulkに足りない5つのこと
2.3.5.は処理の一連の流れを管理する所で解決すべき
Digdag 〜 Workflow Automation System 〜
Workflow Automation Systemとは？
あらゆる手作業の自動化
- データ解析の自動化
- テスト・デプロイの自動化（継続的インテグレーション）
- メール一斉通知
Workflow Automation Systemに求められる機能
- 基本機能
- エラー処理
- 状態監視
- 高速化
- ワークフロー開発
既存のWorkflow Automation System
- OSS
- Makefile
- Jenkins
- Luigi
- Airflow
- Rundeck
- Azkaban
- etc
- 商用
- JP1/AJS3
- etc
- ETLツールも類似のツール
- Talend
- DataSpidar
- etc
Digdagとは？
Digdag is a simple tool that helps you to build, run, schedule, and monitor complex pipelines of tasks. It handles dependency resolution so that tasks run in order or in parallel.
Digdag fits simple replacement of cron, IT operations automation, data analytics batch jobs, machine learning pipelines, and many more by using Directed Acyclic Graphs (DAG) as the infrastructure.
http://www.digdag.io/index.html
Digdagは複雑なタスクのパイプラインのビルド、実行、スケジュール、管理に役立つシンプルなツールです。タスクは順番にまたは並列に実行して、依存関係を解決します。
Digdagは基盤として無閉路有向グラフ（DAG）を使用しており、cron、IT運用の自動化、データ分析のバッチジョブ、機械学習パイプラインなどのシンプルな置き換えに適している。 （筆者訳）
ワークフローエンジンに求められる機能 ＝ Digdagでユーザがシンプルに実現できるようにしたい機能
DEMO: DIGDAG (LOCAL MODE)
CSVファイルをPostgreSQLへロードして、RubyからSQLで集計した結果を、PythonでSlackへ通知。
（会社のSlackを使っているので見ちゃいけないものがあったら目を瞑ってください、との事でしたｗ）
YAMLファイルで『タスク』を定義します。YAMLの中でタスクをグループ化したりなど、依存関係を記述できます。
Operators
- call>: Calls another workflow
- require>: Depends on another workflow
- py>: Python scripts
- rb>: Ruby scripts
- sh>: Shell scripts
- loop>: Repeat tasks
- for_each>: Repeat tasks
- if>: Conditional execution
- fail>: make the workflow failed
- mail>: Sending email
- embulk>: Embulk data transfer
- td>: Treasure Data queries
- td_run>: Treasure Data saved queries
- td_load>: Treasure Data bulk loading
- td_ddl>: Treasure Data operations
- td_table_export>: Treasure Data table export to S3
Operatorをプラグインで簡単に開発できるようになると、自分達のOperatorを作成して、YAMLを書くだけで処理できるようになります。
Commands and Mode
- Local-mode commands
- new
- run
- check
- scheduler
- selfupdate
- Server-mode commands
- server
- Client-mode commands
- start
- retry
- log
- kill
- workflows
- schedules
- backfill
- reschedule
- sessions
- attempts
- tasks
- push
Local-modeはローカルのみ。
Server-modeは、Treasure Dataと同様にdigdag serverをホスティングして使用できます。
将来的にはDockerの上で動いているagentをユーザの環境にインストールできるようになると、どこでもDigdagが動くようになります。
まとめ
- オンラインゲーム・ソーシャルゲームに、データ分析基盤の存在が当たり前に！
- 分析だけじゃない役割も求められるようになってきた。
- データ分析基盤にはデータの流れを作るOSSが大事！ • Fluentd • Embulk • Digdag
- Embulkはバッチでデータ転送の役割をもち、プラグインで色んなデータソースに対応できる ！
- Digdagはバッチ処理のワークフローを簡単に扱うためのオートメーションシステム！
- リンク集はこちら。
リアルタイムサーバー 〜Erlang/OTPで作るPubSubサーバー〜
gumiでR&Dをされている@yamionpさん
なぜリアルタイムサーバーか
これまでのソーシャルゲームは非同期コミュニケーションが中心でしたが、濃い連携や一体感や本当の同時プレイといったより濃密なゲーム体験へ移ってきています。
HTTPの問題
- 通信はユーザー側からしか開始できない
- 高頻度の通信が難しい
TCP Connectionを直接使う
- 最初のConnectionを確立する
- Connectionがある限り自由なタイミングで通信可能
HTTPでもTCPは使っているが
- Responseを返したらConnectionを破棄する
リアルタイム通信のメリット
- どちらからでも送り始められる
- 一回あたりの通信コストが小さい
- 秒間15回とか送っても全然OK
- 遅延が小さい
デメリット
- 負荷が高い・負荷分散しにくい
- 状態を持つので障害に弱い
- エラーパターンが多くコードが複雑になりがち
- 端末のバッテリー消費が激しい
なぜ自作なのか
- Windows Serverは（我々にとって）運用コストが高い
- 機能過多
- 既存システムとの相性
- ゲームでの通信に特有な仕様への対応
- 落としたくない。土日に寝ていられるシステムを
フルメッシュではなくスター型のサーバの話をします。
Pub/Subモデル
Pub/Sub（出版/購読）モデルとは非同期メッセージングモデルの一種
Pub側はSub側を意識する必要がなく、疎結合になる
Roomモデルと違い、Roomの管理が必要ない。
Sub側が0のTopicにもPublishでき、Pub側がTopicをSubscribeしている必要がない。
導入イメージ
プロトコル
概要
- TCPを採用
- TLVベースのプロトコル構造
- 最小限の固定ヘッダと拡張部分のTLV化
- 4 bytes 単位
この構造によるメリット
- 拡張しやすい
- 実装が比較的容易
- バリデーションが容易
RTTを計測する
- ユーザ⇔サーバーの回線状態を把握したい
- 強制切断など
- 通信時に自分のタイマーと処理時間を加算した相手のタイマーを送る
なぜTCPか？
- NAT超え
- 到達保証
- 順序保証
- 輻輳制御
- ゲームによってはTCPのみでも十分な場合も多い
UDP拡張も試験中
サーバー
コンセプトは
- 土管に徹する
- ステートフル、データレス
- 品質重視
なぜErlang/OTPなのか
Erlang/OTPとは
- エリクソン社製の関数型言語
- 動的型付け − 優れた耐障害性
- 並行処理・分散処理に強い
軽量プロセス
- 起動速度・使用メモリが超軽量なプロセス
起動時間は25μ秒、使用メモリはプロセスあたり3kB1 - インスタンスを作る感覚でプロセスを作る
- プロセス間ではメッセージ通信ができる
Let it crash
- エラーが起こったプロセスは死ぬ
- プロセスが死んだことは監視している別のプロセスに対処させる
- 再起動や後始末をしたり、連携する他のプロセスの処理をしたり
- 監視プロセスは別マシンで動いていても問題ない
- 正常系と異常系の処理が分離される
負荷試験
なぜ行うのか
- 限界性能を知る
- ボトルネックの発見
- スケールすることの確認
- 必要容量の計算
- コスト試算
Locustを使用して負荷試験を実施
1msgを7デバイスへ送信する事（1room辺り8ユーザ）を想定
UDPのスループットがTCPの1/5問題
- V「5万/sはいける」
- 1万/sでない。
fprofでプロファイリング
- sendが遅い
- port_commandで非同期にしたら解決
- gen_udp/tcp:send がボトルネックなときにやること
- 10万/s 出ました
実際のゲームでの使われ方
- クライアントからAppサーバへマッチングを依頼すると、Appサーバがマッチングのリストへ追加する。
- Appサーバはクライアントへ、マッチングの成功可否を返すPub/Subサーバのポートを返す。
- マッチングサーバがリストを監視しており、組み合わせを決める。
- 組み合わせを決めると、現在組み合わせの全員が待っているかを問い合わせる（Private Topicへメッセージを送る）。
- 全員が開始できる場合は、返事専用のTopicへ返す。
- 全員がOKの場合にAppサーバでゲーム開始の処理を行う。
- クライアントはゲームの情報をAppサーバへ貰いに行く
HW障害への対応
作ってみて
- 思ったより速度が出た
- デグレさせないことの重要さ
- 運用を最優先の姿勢
One More Thing
ArkPS ServerをOSSとして公開予定です。
https://github.com/arkps/ （予定地）
リアルタイム通信ゲームの実装例
オンラインゲームを作るのに必要になる通信技術のお話
- 通信規格
- 通信内容
- 通信ライブラリ
- RPC
- IDL
通信規格
- UDP： 単方向通信
- TCP： 双方向通信
常時接続 / 非常時接続
- 非常時接続（単方向通信）
- リクエストに対してリプライ
- クライアントからの要求が必須
- 常時接続（双方向通信）
- リスエストに対してリプライ
- サーバ側からのプッシュが可能
- クライアントに要求できる
通信内容
- キーフレーム形式
- メッセージ形式
- ゲーム、プロジェクト規模に合わせて適切な規格・形式を選ぶ
メッセージ / キーフレーム
- 通信量の少なさ
- キーフレーム ＞ メッセージ
- 処理の簡単さ（高速）
- キーフレーム ＞ メッセージ
- 拡張の容易さ
- メッセージ ＞ キーフレーム
- 複雑なデータのやり取りのし易さ
- メッセージ ＞ キーフレーム
通信ライブラリ
メッセージ形式を利用したライブラリ
- ソケット部分のコード
- RPCの提供
- IDLの提供
RPC (Remote Procedure Call)
- プログラムから別のマシンで動いているサブルーチン（手続き）を呼び出す仕組み
- 通常はネットワーク越しに、通信対象のマシンで動作するプログラムの手続きを呼び出す
実現するためには
- RPCを実現する為に必要になりそうな情報
- 呼び出し順序の保障
- 適切な単位（メッセージ毎）の受信
- バイトオーダーの吸収
環境によってデータの表現（エンディアン）が異なる
パケットの構造
- RPCを実現する為に必要になる通信内容
- プロトコルバージョン番号
- 制御用カウンタ
- どの関数を呼ぶ？
- 関数に合わせた引数
- １手続きを実行するのに必要なデータ総容量
レイヤー構造による処理
実装における問題点
- 通信内容を一つ追加する都度、追加・修正しないといけないコード量が多い
- 送信関数
- 受信関数
- 受信関数の呼び分け部分（Dispatcher)
- これを簡易化する為に IDL（Interface Definition Language） を用意する
IDL
- IDL（Interface Definition Language）
- この定義を元に、最低限必要になるコー ドを自動生成する仕組み
- 定義ファイルとコードジェネレータ
- 送信側
- 受信側が識別可能な一意なメッセージID
- メッセージ毎のパラメータ格納
- 受信側
- どのメッセージを受信したのか？（適切な手 続きの呼び出し）
- メッセージ毎のパラメータ展開
- 受信後の処理はアプリケーションレベルの実装
メッセージ定義を元にGeneratorが『サーバ用 受信』と『クライアント用 送信』のコードを各種言語で自動生成します。
まとめ
- 通信方式（TCP/UDP/常時/非常時）
- 通信内容（メッセージ/キーフレーム）
- 通信ライブラリ
- RPC
- データの表現方法（バイトオーダー）
- メッセージの表現（パケット構造）
- メッセージを処理するレイヤー構造
- IDL
- IDLの例
- RPC
そして懇親会
懇親会と言ったらBeer & Pizzaですよね！皆様との会話で発表内容の理解が進みました。
まとめ
Digdagに始まり、リアルタイムサーバ実現のためのプロトコル実装まであり理解が追いつかず非常に中身の濃い勉強会でした。
DigdagはOSSでAgent方式で構成するワークフローエンジンと言うことで、個人的に非常に注目していました。またリアルタイムサーバのためのプロトコルを自前で実装するといった、触れないレイヤーのお話は非常に勉強になりました。
Digdagはコードが公開されていますし、ArkPS Serverは今後OSSとして公開予定と言うことで今後が非常に楽しみです。
27
