CloudのRabbitMQ

最近PaaS上でOSSがいろいろ使えるようになってます。
ちょっとQueueを使う機会があったので、
せっかくだからとrabbitMQ on Bluemixを
SparkStreamingとかSpark処理のoutput先として使おうと思った。らハマった。

Queueなんで必要？

昔からシステム間連携の際のプロトコル変換などで利用されてました。
最近は単一のミドルウェアやサーバーは(HW障害などで)落ちる前提で設計することが多いですが、SPOFの排除とシステム全体での可用性担保のために使ったり、従来でも使われてたシステムを疎結合に保つために使われたりとかしてます。

マイクロサービス間でのメッセージの保証
非同期にDBとかファイルストレージとかデータストアに書き込む際のデータの保証
非同期にブラウザへ処理を返す所とか

今回データ分析でSpark処理の出力先としてQueueを使ったのは
時間がかかる機械学習のモデル作成やデータ更新を受け取り側でポーリングしたくなかったから。ちなみにデータの入り口部分を扱うならKafkaの方が向いてます。RabbitMQ使うなら

Publisher > Consumer

とデータ量がConsumer側で少なくなる場所で使うと良い仕事してくれます。
逆に

Publisher ≦ Consumer

となるケースでは基本的にRabbitMQはPush型なのでConsumerが詰まるケースで取り扱いに注意。特にオンプレでRabbitMQ構築する場合にはHighWaterMark超えてDiskにデータが落ちるとConsumer側のパフォーマンスが著しく劣化するので。
大量のデータ扱うならKafka、即応性でRabbitMQという感じで使い分けてます。

どんな感じで使ったか

Sparkの分析処理の結果を

Spark処理の方ではrddをcollectしてDriverからRabbitに結果を投げる感じ。
Streaming処理の方ではforeachRDDの中で投げる感じで。
RabbitMQに一旦Queueingして、Node.jsでConsume。
そのままWebsocketでブラウザに分析結果を垂れ流し

的な感じで利用してます。以下、絵でのイメージ。

CloudのRabbitMQ

Bluemixのアカウント持ってればボタン１個で出てきます。構築は以上。

払い出すと以下の画像のような管理ページに飛ばされます。
サービス資格情報のタブの中にエンドポイントやユーザー情報が入ってるので、それを使ってRabbitMQにデータ入れたり出したりします。

ちなみに
Note: Compose via Bluemix does not give access to the "Compose UI" at this time, see here for more details.
なる注意書きがありますが、普通にRabbitMQの管理画面が使えるので
(ﾟεﾟ)ｷﾆｼﾅｲ!!

サービス資格情報タブから、uri_adminの項目のURLで管理画面にアクセス可能です。ID/Passwordはuri_direct_1のbasicの所のadminと後ろのパスワード使います。(パスワードは黒塗りしてる所)

おなじみのRabbitMQの管理画面です。

ハマった所

今回PySparkからこのRabbitMQにデータを投げ込みたかったのですが、
上のサービス資格情報のイメージのuri_direct_1の項目見ていただいたらわかる通り、
amqpじゃなくてamqpsでのアクセスとなります。
本家にPythonでpikaを使ったサンプルありますが、このままエンドポイントだけ変えて写経すると処理がこけます。

普段Queueってネットワーク内部に置くことが多いのでamqpsって使わなかったんですが
当たり前といえば当たり前ですが、ちゃんとSSLOptionつけたげないとダメみたいです。サービス資格情報タブのca_certificate_base64をURLの後ろにくっつけてあげるときちんと処理が通ります。

意外と忘れがちなので。備忘的に。

同様にconsumer側もssl設定する必要があります。

Cloud上でいろんなOSS出てますが、サクッと環境作れるのは楽でした。
特に環境構築に時間使いたくない場合は是非。