<repositories>
    <repository>
        <id>koduki-repos</id>
        <url>https://raw.githubusercontent.com/koduki/koduki.github.io/mvn-repo/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>cn.orz.pascal</groupId>
        <artifactId>jl2</artifactId>
        <version>0.1-SNAPSHOT</version>
    </dependency>
</dependencies>

Tuple

全ての基本となるTupleです。まあ、ほぼScalaからの移植です。下記のように使います。

Tuple2<String, Integer> x = new Tuple2<>("A", 10);
Tuple2<Integer, String> y = $(1, "B"); // '$' はシンタックスシュガー
Tuple4<Integer, String, Boolean, List<Integer>> x = $(1, "A", true, Arrays.asList(2)); // Mapと違っていくらでも型が書ける
Tuple2<Integer, Tuple2<String, String>> z = $(1, $("A", "B")); // 入れ子もOK

配列と違って複数の型を管理できます。$関数を一部使ってますが、これはnew TupleXのシンタックスシュガーです。

これが使えるだけでStreamAPIをはじめ使い勝手が随分良くなります。

コレクションビルダー

これは特に関数型は関係無いですが、Javaのコレクションは初期化が面倒なのでScalaっぽい記法のシンタックスシュガーを作りました。

Set<String> set = set("a", "b");
List<String> list = list("a", "b");
Map<Integer, String> map = map($(1, "a"), $(2, "b"));

特に、MapはTupleの組み合わせることで通常より格段と書きやすくなってると思います。

Immutable なコレクション操作

たまにJavaで再帰を書こうとするとImmutable なコレクション操作ができないので直感的に書きづらかったりします。

なので下記のように簡単に書けるライブラリを作りました。

まあ、この辺はより本格的なライブラリがGuavaとかにあると思うので、そっちの方がより便利に使えると思います。

Set<Integer> set1 = set(1, 2);
Set<Integer> set2 = set(3);
Set<Integer> concatedSet = concat(set1, set2);

// create immutable map
ImmutableMap<String, String> map = imap($("k1", "v1"), $("k2", "v2"));
ImmutableMap<String, String> addedMap = map.put("k3", "v3");
ImmutableMap<String, String> removedMap = map.remove("k2");

JDBC Wrapper for StreamAPI

プロトタイプやシンプルなアプリの場合は、ORM使わずに直にJDBC使うこともそれなりにあると思います。

ただ、ResultSetはStreamAPIに対応してないので書くのが少し面倒。というわけで、対応させてみました。

try (
        Connection con = DriverManager.getConnection(JDBC_URL, USER, PASSWORD);
        PreparedStatement pt = con.prepareStatement("select * from persons ");) {
    List<String> ids = QueryStream.of(pt.executeQuery())
            .map(r -> r.getString(1))
            .collect(Collectors.toList());

    assertThat(ids.size(), is(4));
    assertThat(ids.get(0), is("1"));
    assertThat(ids.get(1), is("2"));
    assertThat(ids.get(2), is("3"));
    assertThat(ids.get(3), is("4"));
}

JDBCをもうちょっとモダンに使いたいけど、大げさなライブラリを入れたく無い時に便利です。個人的にはTupleとMapの初期化に次いで良く使う機能です。

まとめ

本格的な関数型サポートを行うライブラリは色々ありますが、そこまで大げさなものは要らないという時に、SFライブラリのJL2はいかがでしょうか？

それでは、Happy Hacking!

2017-07-20

分散システムの「キホン」の「キ」 - あるいは普通のWebアプリの作り方

読み物

みなさん、分散システムというと何を思い浮かべますか？ Hadoopですか？ Sparkですか？

現代だと多くの人がそういったイメージを持つと思いますが、調べてみるとそれらは厳密には分散「コンピューティング」に属する技術のようです。

分散システムとはメインフレームの対義語つまり普通に我々が今使ってるほとんどのコンピュータですね。

分散システム - Wikipedia

当たり前すぎてふだんは「分散」って部分を忘れがちなんですが、これを意識してシステム開発をしないと思わぬところで落とし穴にはまります。

という分けで、今回はシステム開発をするにあたって、分散システム—つまり普通のWebアプリのようなシステムを組む際に、どういう所が問題になりやすいかを考えてみました。

分散システムの特徴
基本的な対処方法
まとめ

分散システムの特徴

たくさんの処理を同時に実行するのに向いてる

分散システム、すなわち現代的なコンピューティングは「同時」に「複数の事」を処理するのにとても向いています。

たとえばRailsやJavaEEのようなWebアプリケーション。秒間100件のスループットを考えたときに「0.01秒間の処理を100回実行するシステム」と「1秒の処理を10個同時に実行するシステム」なら断然後者の方が簡単に作れます。

特に最近は「スケールアップ」と言っても実際はCPUのコア数が増えるだけのこともありますし、台数を増やすことでスケールアウトするのも一般的です。例外はいっぱいありますが。

なので、基本的には「超速い処理」を作るよりは「まあまあの速度を大量に実行する」方向にミドルウェア等も進化してるので、それを意識した設計をすることが大事です。

早く到着したからと言って、最初に発生した順とは限らない

では、同時に処理を実行したさいの弊害はなんでしょうか？　その一つは順番が狂う事。つまりイン・オーダー実行が保証しきれないことです。

ネットワークの遅延だったり、負荷分散システムのバランシングの都合だったり、様々な理由で順番がずれます。

これを考慮せずにシステムを組むと… チケットの予約よりキャンセル処理が先に来たりなかなか楽しいシステムができます。

ほとんど同じ時間に処理されることもある

上記の話と少し似てますが、ほとんど同じ時間に処理されることもあります。ミリ秒以下の単位で。

これで何が困るかというと雑な乱数やタイムスタンプをキー情報にするとキーが重複するってことです。

乱数のデフォルトシードがタイムスタンプな言語は多いと思うので、乱数を使ってるつもりで重複への品質がタイムスタンプと同程度だったというのは良くある笑えない話。

排他制御はあるけれど…

他にもデータの更新処理は要注意です。二重更新問題によるデータ破壊が起こります。

できるエンジニアになる！ちょい上DB術・基礎編（2）：【DB概論】DBMSに求められるもの（1）排他制御とACID属性 - ＠IT

そういったことを防ぐお馴染みの方法の一つが上記のページに記載さているような「ロックによる排他制御」です。

排他制御は言ってしまえば並列処理をシリアライズしてシーケンシャルに処理するので、並列処理にまつわることが大体解決しますが当然性能が落ちます。

特に、カウンターとか合計値の計算とかはロック待ちになり、台数を増やしてもマシンリソースが余っていても全然性能が伸びないってことになりがちです。

基本的な対処方法

では、上記の特徴を踏まえて基本的な対策というかシステムの作り方を紹介していきます。

下記のような構成のシステムをベースに考えていきます。

|Front App| -> (バランサ) -> |Back-end API| -> |RDB|

到着順の問題はどう解決すればいい？

到着順とイベントの発生順がずれる問題はいくつか対処方法がありますが、ひとつはイベントが発生した時間(Event Time)を値として持って置き、少し間をおいてソートしなおして実行することです。

この「少し」って範囲はシステムごとに異なります。1秒程度なのか1分なのか1時間なのか1日なのか、業務的に許容できることろは異なってくるでしょう。メモリでもRDBでも一時的に格納してしまい、ウインドウ集計をすることで問題を回避できます。

たとえば、「登録」と「キャンセル」が非常に頻繁に行われる場合はBack-end APIでは到着順に「予約管理一時テーブル」にデータを突っ込むと実際のタイムスタンプと異なるイベントタイムで格納します。

この際重要なのはポイントは何らかの方法で「イベントタイム」をパラメータに含めておくこと。

予約管理一時テーブル：

予約ID	ユーザID	操作	Timestamp	Event Time
1	Shiro	キャンセル	00:01	00:02
1	Shiro	登録	00:02	00:01
2	Emiya	登録	00:03	00:02
3	Emiya	キャンセル	00:04	00:03
4	Bob	登録	00:05	00:05

この状態で予約管理一時テーブルから、「例えば5秒毎に予約IDで抽出してEventTimeでSortして上から順に結果を適用した結果を予約管理テーブルに格納する」みたいな処理を作れば、到着順がばらけてしまうようなケースでも不整合なく処理が出来ます。

予約管理テーブル：

ユーザID	ステータス	Timestamp	Event Time
Shiro	キャンセル	00:06	00:02
Emiya	登録	00:07	00:03
Bob	キャンセル	00:10	00:05

また、ユーザや別のシステムは「予約管理テーブル」を見る事で5秒遅延した状態ながらも正確な状態を見ることが出来ます。

閲覧者にとって厳密なリアルタイムのステータス更新が必須でなければ十分にとれるアプローチですね。

また、難易度は上がりますが、5分の枠を1分ずつずらすとかでウインドウを重ねて集計すれば精度はかなり高くできます。

それでも業務上許容できないときは、例えばFront App側がBack-end APIに投げる時に同期処理として順番に投げるとかクライアント側での工夫が不可欠になります。

ただし、性能を犠牲にせざる得ないので、ウインドウ集計で作って万一問題が出る場合は、別途整合性チェックの処理を走らせて業務的に回避するとかバッチで再計算するとかが出来れば検討したい。

被らないキーはどう作る？

さて、分散システムで以外にめんどくさいのがユニークキーの生成です。

一見するとタイムスタンプで良さそうですが、単純なタイムスタンプでは十分に並列度があるとわりと被ります。特に精度がミリ秒だともう普通に被る。

なので、別な方法を考える必要があります。一つはRDBのシーケンスなどを使ってユニークキーを作る方法。これはRDBを利用したシステムだと非常に手軽なのでIDとかに使いやすいですね。

ただし、シーケンスの発行がRDB側でボトルネックになる可能性が大いにあります。Oracle RACであればCache + NOORDERである程度改善出来ますし、他のDBでも似たような改善方法はあると思われます。

それでも、一カ所でコントロールというのは限度がありますし、シャーディングしたDBに一意に書きたいこともあるので、そういったときはアプリ側で実装します。

独自で組む(ナノ秒 + メモリサイズをシードとした乱数 + スレッドID + MACアドレスの組み合わせとか)
UUIDやSnowflakeなどの既存の仕組みを使う

1のように独自で組むというのも一つの手です。環境や状況が搾れるので比較的軽量なロジックを組むことが出来る可能性があります。また、2で記載してるように分散環境でもユニークキーが求めれる仕組みを使うのも手です。

特にUUIDは色々な言語での実装がすでにあると思うので、導入が手軽なのは魅力的です。

RDBでのUPDATE文でのロック待ち対策

RDBでUPDATE文で更新が重なるとロック待ちになるのは仕方ないです。これを改善することは出来ません。なので、UPDATE自体を極力減らしロック対象も小さくする事で「ロック待ちが重なる」という事そのものを減らすのが基本戦略です。

正規化しよう

まずは、ロック対象を減らす話。

カラムを大量にもったテーブルの場合、ロックが必要なカラムがアプリによって違う場合もあります。これは単純にテーブルの設計が悪いですね？

なので正規化を検討しましょう。正規化をしてロック対象を小さくすることで、ロック待ちが競合する可能性を減らすことが出来ます。

たとえば、以下のようなテーブルを考えます。

ユーザ：

名前	フレンド件数	サーバント数
koduki	30	90

上記のような作りにしてしまうと、フレンド申請とガチャが被ると同じ行に更新ロックが掛かってしまいます。

なので、下記のようなテーブルとします。

ユーザ：

ID	名前
1	koduki

フレンド数：

ユーザID	フレンド数
1	30

サーバント数：

ユーザID	サーバント数
1	90

こんな感じで、3つのテーブルに分けることでフレンド申請とガチャで更新ロックが被ることは無くなります。

UPDATEからINSERTに切り替える

次に検討するべきはUPDATE文からINSERT文に切り替えることです。

そもそも「UPDATE文でロック待ちになる」ということは何等かの共通データであり、多くの場合それは「合計値」や「件数」といったサマリテーブルです。

このようなテーブルは下記のようにINSERTベースのテーブルにして、SELECT時に集計するという方法が考えられます。

たとえば、以下のようなテーブルを考えます。

フレンドに呼ばれた回数一覧：

名前	回数
ステンノ	3
エウリュアレ	5
メデューサ	1

この場合、参照時のSQLはシンプルに

select * from フレンドに呼ばれた回数一覧;

ですね。ただ、これだと同時にフレンドから呼ばれた場合はロック待ちが発生していします。人気サーヴァントだと相当時間がかかってしまいますね。

なので下記のようにします。

フレンドに呼ばれた履歴一覧：

名前
ステンノ
ステンノ
ステンノ
エウリュアレ
エウリュアレ
エウリュアレ
エウリュアレ
エウリュアレ
エウリュアレ
メデューサ

この場合、参照時のSQLは下記のようになります。

select 名前, count(名前) as 回数 from フレンドに呼ばれた履歴一覧 group by 名前;

いずれも、得られる結果は

名前	回数
ステンノ	3
エウリュアレ	5
メデューサ	1

なので、業務的には等価ですね？ INSERT方式の場合はロックが発生しません。*1

このように実際に参照するモデルとは違う格納モデルにして、性能を稼ぐことが考えられます。

ただ、この作りは性能が出ないケースがあります。最近のDBは十分に速いので意外とそのまま行けるケースも多そうですが、性能が出ない場合は下記を検討します。

Materialized View
サマリテーブル + (Trigger or バッチ)
(1 or 2) + 直近データだけ集計して合計する

1, 2の方法はgroup_byの計算結果をデータとして保持しておくパターンです。書き込みと参照で若干の不整合が許されるなら、これだけで十分です。

厳密な整合性が必要な場合は、3の直近の1分間とか1時間とか性能にインパクトを与えない範囲のデータだけトランザクションテーブルから集計して、サマリテーブルの値と合計する方式になると思います。

システムとしてはちょっと複雑になっちゃいますが、この方法であれば厳密性と並列性(性能)を両立できる可能性があります。

まとめ

分散システムという事に注目して、ふつうにWebアプリ等を書くときにスケールしなくなるポイントを整理してみました。

スケールさせるにあたって非同期キュー入れたりするケースとかも多々あるでしょうが、そもそも根底としては今回書いたようなケースを意識する必要があるんじゃないかと。

まあ、あんまり課題と例が適切じゃない気もするけど、そこはイメージで（ぉ

他にも色々あるでしょうし、「それは違うよ！」ってのがあればご指摘/コメントいただければと。

それでは、Happy Hacking!

*1:索引のロックが掛かるケースはあるのでRDBに応じてパーティション化等が必要になる事も

2017-07-11

Hadoop の時代は終わってないけど、使いどころは限定されてきたかもしれない

読み物分散

id:shiumachi さんが書かれてる下記の記事がとても良かったです。 shiumachi.hatenablog.com

私自身もSparkを触る前は「Hadoop == MapReduce」と思ってましたが、どちらかというとYARNやHDFSがHadoopファミリの核だと最近は思いますしその意味でのHadoopは全然終わってないと思います。記事の中で書かれてる通り、ある意味ではさらなる進化を遂げて花開いてる状態かな、と。

ただ、Twitterにも少し書いたんですが一方で「猫も杓子もHadoop！」の時代が終わりつつあるのも事実かな、と思います。

もうちょっというとHadoopに限らず巨大スケールの分散システムの用途が収斂してきたのかな、と。

HadoopやGoogle MapReduce登場時のと違い、ストレージI/Oは133MB/sとかの単位で争ってたHDDからストレージに代わって、今や3GB/sを超えます。しかもランダムアクセスもHDDよりずっと速くなっている。

ちょうど最近発表されたZenベースのサーバだと64コア/128スレッドで、メモリも4TBです。普通に当時の10倍以上のマシンスペックがある訳です。単純に。

pc.watch.impress.co.jp

また、そういったハードウェアスペックに合わせてインメモリDBやクラスストレージメモリ、あるいはGPUを活用する形にRDBなどのミドルウェアも進化しています。

それこそ2010年からしばらくは「デカいデータ処理（あえてビックデータとは呼ばない）は分散システムでやるべし、すなわちHadoop！」って感じだったと思っています。

これは本来Hadoopがターゲットにしているビックデータ（TB級やPB級）ではなく数百GB級でも普通に組んだら性能足らなくて仕方なく使ってたケースがあるかと。

そういった用途で使われていたHadoopファミリは徐々に利用されない形になっていくんじゃないかと。

例えば、3GB/sのI/O性能が出るなら、1TBのデータ読み込みですら5分と少しです。

数TBのサイズであればインメモリDBにデータがすべて乗ります。

数十から百程度の並列処理なら分散しなくても単体マシンで処理可能です。

こうなってくると機械学習や特大のDWH以外はTB,PBクラスの処理をすることは少ないと思います。特に歴史がある会社だと昔から運用してるので業務的に圧縮してるはずだから、通常業務のデータはそんなにでかくないはずなんですね。日次とか月次でサマったり。じゃないと元々運用できないですし。

なので、そういった業務に対して単純に利用者が増えるとかの線形なデータ増加に対する対応だとHadoopのような巨大な分散システムを運用する必要性は小さくなってきます。

もちろん、冗長性やローリングアップデートのためにもシステムが多重化されてる事は大事なんですが、数百台のクラスタで動かすことを想定した分散システムではなく、それなりに信頼できる数台のマシンで構成した小さなクラスタを運用する方向性が強まってくるんじゃないかな、と。

ただ、強まってくるんじゃないかと思いつつもRDBに分投げる方はともかく、分散処理エンジンであるSparkに代替するような並列処理エンジン/管理システムがあんまり無いのも困りもの。

自分が知る限りだとそのコンセプトを強く打ち出してるのはAsakusa-on-M3BPくらいですね。

Asakusa on M3BP — Asakusa on M3BP 0.1.3 documentation

RDBをインメモリで運用できてそっちに処理をオフロード出来るケースなら、あまり問題ないのですがそうじゃない場合は「複数スレッドをマネージするバッチの処理基盤」自体はやはり必要です。

何も考えずに各々のアプリが好きなサイズのスレッドを取得して使うと、ジョブスケジューリングが適切にできなくなってしまいます。

なので、Asakusaのような思想でのミドルがもう少し増えて、運用ノウハウが溜まっていけばな、と思ってます。

あるいはMesosやk8sでリソースを別途管理する基盤を作るかですね。JavaEEのjBatchもそういう方向で活用が出来そうだけど、ぶっちゃけ登場が遅すぎて機能とエコシステムが弱いのがネックですねぇ。

この辺は決定版が無さそうだから管理ツールとか含めて自作する必要があるかも。

まとめ

なんか途中から大分脱線した気がしなくもないけど、Hadoopの時代は終わってなものの「本当はHadoop使うには微妙だけど他に選択しないからHadoop」って用途がHWやMWの進化で、適切なシステムに移行していくので、結果としてHaoopのような大規模分散クラスタは特定の用途に寄るだろうなってのが言いたい事でした。

それではHappy Hacking!