CLI で AWS Kinesis Stream の中身をひたすら追ってくれる k-iter をつくってる
作ったもの
インストール方法
rustup.rs - The Rust toolchain installer
rustup を設定したうえで、下記コマンドを実行。
cargo install --git https://github.com/ROki1988/k-iter.git
もしくは Releases · ROki1988/k-iter より合致するものをダウンロードして展開。
使い方
k-iter -n event-stream -r ap-northeast-1
n
の後にストリーム名を、r
の後にリージョン名を入れればガンガン見てくれます。
今のところはコマンド実行以降に Put されたレコードを見ていくだけです。
動機
AWS Kinesis Stream はいったん投入してしまうと、中になにが入ってるか追うためにコードをかく必要があります。コンソールから見れないから。
Lambda の Blueprint に Kinesis のイベントを処理するぜー、てのもあるのですが、CLI に出したい、という欲がでます。
ですので、秘伝のたれのごとき Python コードを実行して CLI 上で出してました。
だがしかし、迂闊に brew upgrade
を実行した結果、Python の実行環境が再構築の憂き目にあい、こりゃシングルバイナリで動くやつ作らにゃきつい、となった次第です。
Rust 製なのは趣味です。
やってみたいこと/考えてること/もらえた意見
- 実行バイナリはよ
- Verbose モードとかほしい。Put された時間とかパーティションキーとか
- Iterator Type 対応したい。
TRIM_HORIZON
とか。- オプションに依存関係が出てくるので、ちょいと厄介
clap-rs の修行が必要そう
- オプションに依存関係が出てくるので、ちょいと厄介
--exec
とか作ってデータに処理かけたいjq
かけたりするイメージ。やりたくない?
- UTF-8 文字列決め打ちで出力してるけど、バイナリ表示の需要もありそう?
- Big とか Little の切り替えもいるのだろうか。
--print-format=string|byte-be|byte-le
てな感じ
- Big とか Little の切り替えもいるのだろうか。
- Shard 複数を同時に
- できっかなぁ…
Rust で Scala の continually ぽいもの
こんな感じ?
extern crate rand; use rand::{Rng, thread_rng}; use std::usize; pub struct Iterate<A> { func: A } impl<B, A> Iterator for Iterate<A> where A: FnMut() -> B { type Item = B; #[inline] fn next(&mut self) -> Option<B> { Some((self.func)()) } #[inline] fn size_hint(&self) -> (usize, Option<usize>) { (usize::MAX, None) } } fn continually <A, B>(f: A) -> Iterate<A> where A: FnMut() -> B { Iterate { func: f } } fn main() { let mut rng = thread_rng(); let a = std::iter::repeat(1).into_iter().map(|_| rng.gen::<u32>() % 100u32).take_while(|x| *x != 0u32).collect::<Vec<u32>>(); let i = continually(|| rng.gen::<u32>() % 100u32).take_while(|x| x != &0u32).collect::<Vec<u32>>(); println!("{:?}", a); println!("{:?}", i); }
きっかけ
Iterator.continually はそういうことかー、と納得してる
— 萬屋 rohki (@r_ohki) 2018年4月12日
Scala の continually をみて、なるほど都度評価してくれるのか、と納得して、はて Rust では、となった次第。
書いてみて
std::iter::repeat
もあるにはあるのですが、Clone
trait を実装している必要があって、FnMut
を渡せませんでした。たぶん。
ので、軽く書いてみたら動きました。でもありそう。見つけられてないだけで。
あと、表面だけまねてるので、いろいろ考えが足りてなさそうでもあります。
でも動いたから満足。
Elasticsearch のクエリを知るのに elastic4s がよかった
Elasticsearch はややこしい
ややこしいというか難しいというか、面倒くさいというか…
で考えたところ、そも JSON を組み立てるのがそこそこ以上にしんどいんですよね。末尾の ,
とか。
Kibana も補完はあるのですが、そのあたりが面倒くさくて、なんか学ぶのが億劫になっておりました。
で、表題の話。elastic4s です。
sksamuel/elastic4s: Elasticsearch Scala Client - Non Blocking, Type Safe, HTTP, REST API, TCP
Scala で Elasticsearch を扱うためのライブラリになってます。
かなーり頑張っていて、検索結果を safeTo で変換できたりとか、ややこしい aggs
に型の誘導があったりします。
あと 6.0 以上であれば AWS Elasticsearch Service の IAM 認証付きリクエストも行けます。AWS_DEFAULT_REGION
を環境変数にいれなきゃですが。
Aggregations がすごい
いやーえぐい。ほんとえぐい。というかこれだけの量をよく定義しはりました。
val resp = http.execute { search("childrenaggs").matchAllQuery().aggs( dateHistogramAgg("agg1", "date").interval(DateHistogramInterval.Month).addSubagg( childrenAggregation("agg2", "answer").addSubagg( termsAgg("agg3", "text").size(1) ) ) ) }.await.result val september = resp.aggs.dateHistogram("agg1").buckets.find(_.date == "01/09/2017").get val sept_answers = september.children("agg2") sept_answers.docCount shouldBe 3 sept_answers.terms("agg3").buckets.head.key shouldBe "god"
テストコード から抜粋したのが上記です。
入れ子の入れ子の入れ子、になってるのかな? もはやわからんわけですが、型に沿って書けるのでよかったです。
んで、aggs
で返ってくる結果のキー名とか方とかも決まってるので、これもとれます。
テスト支援のモジュールもある
まだきちんと試せてないですが、テスト用のモジュールもあります。
Docker を起動するタイプとか、組み込みタイプとか。至れり尽くせり。
サンプルもあって試せる
elastic4s/samples at 151fea1fb680bed793bd3c89149b266bcb9c6129 · sksamuel/elastic4s
丁寧なことに Maven やら Sbt のサンプルプロジェクトも用意されてるので、結構手軽にためせます。
ここでクエリを書いて、.show
で文字列にして確認してを繰り返して、理解が深まった感じです。
与太話
ElasticDsl.scala ってファイルを見ながら、確かに技術ドメインであってもドメインであって、表現してくれてる言語があれば理解が深まる、ってことなのかなー、ORM とかもそういう役割あるのかなー、などと思考が飛躍しておりました。
与太話おわり!elastic4s よいすよ!!
Scala で Array 等を制限に合わせて分割する
短く
grouped(size)
を使いましょう
背景
Scala で AWS Kinesis へ PutRecords しようとしたときに、上限に引っ掛かりました。
で今回抵触した上限というのが、PutRecords 1 回あたりに含められるレコード数で、500 までとのこと。*1
なので 500 毎に分割して PutRecords したいわけだけど、分割…あるはずだよなーと探して見つかりました。
方法
(1 to 10000) .grouped(500) // これ .foreach(put)
たったこれだけです。
ただまぁ自前で作ると脳みその裏側でいろいろ考えたりするわけですよ。
そも動くようにするところから始まり、速度とか、オブジェクト生成の回数とか、いろんな Collections への対応とか。
あるなら使うがベストです。こんなの絶対にあるはずですし
おわりに
名前がよくない!(理不尽)
splitAt
は違うしなぁとかで引っかかってました。
sliding
まで行きついてコード書いた後に、あれよく見たら grouped
あるやん、でやっとこさ到達しました。
Rust では chunk
で、そうか chunk
というワードがあったか、という感じ。検索のための語彙が増えました。
Prometheus + Grafana + Elasticsearch + Kibana を Docker でいじり中
いじってます。
進捗
docker-compose.yml
はこんな感じ
version: '3' services: prometheus: image: prom/prometheus container_name: prometheus volumes: - .\prometheus.yml:/var/app/prometheus/prometheus.yml command: - '--config.file=/var/app/prometheus/prometheus.yml' ports: - 9090:9090 depends_on: - elasticsearch_exporter links: - elasticsearch_exporter grafana: image: grafana/grafana container_name: grafana ports: - 3000:3000 env_file: - /prom/grafana.env elasticsearch_exporter: image: justwatch/elasticsearch_exporter:1.0.2 command: - '-es.uri=http://elasticsearch:9200' restart: always ports: - 9108:9108 depends_on: - elasticsearch links: - elasticsearch elasticsearch: image: elasticsearch ports: - 9200:9200 kibana: image: kibana ports: - 5601:5601 depends_on: - elasticsearch
prometheus.yml
global: scrape_interval: 15s evaluation_interval: 15s external_labels: monitor: 'codelab-monitor' rule_files: # - "/var/app/prometheus/alert.rules" scrape_configs: - job_name: 'prometheus' static_configs: - targets: - '127.0.0.1:9090' - job_name: 'elasticsearch-exporter' static_configs: - targets: - 'elasticsearch_exporter:9108'
Prometheus + Grafana を Mac 上で docker-compose で起動 - Qiita
上記の記事をベースに、Elasticsearch + Kibana を追加して、Elasticsearch 用の Exporter も入れました。
Docker でのアドレス解決がわからなかったので、そこはこちらを読みながらなるほどー、と書いてみました。
Windows でやったので、一部ファイルがおかしいかも…
メトリクス周りを前から調べていて、Graphite や Influx、Mackerel もさわったし、はやりの Prometheus もやってみよう、という感じ。
去年の 2.0 出たときにいじりたかったけど、まぁほかに触るものがあれやこれやそれやあり遅れました。
Elasticsearch で GC に関連した値もとれるってのが個人的には気になってます。exporter よいなー。
一方で、いろいろ手広くやりすぎててまずいです。*1
ここのところ毎年いってるけど来年度は絞る。
*1:僕のせいだけではない
elastic4s の json ライブラリ切り替えの実装にちょっと感動してる話
いつかの続き
elastic4s/package.scala at master · sksamuel/elastic4s · GitHub
で、件のコードがこの辺り。ひっじょうに薄くて、なんと 1 ファイルだけだったりします。
以前フォーマッター切り替えについて考えた記事を書きましたが、さらに先にある実現方法がこれか! と感動しております。
そうだよなー。中間の型クラス作って、委譲すればできるよな。
Kinesis Firehose のデータ変換 Lambda を Golang で作って計測してみた
概要
前回 やるといってたやつです。
AWS Lambda が Golang を公式サポートしたので、これまでやってきたのと同じような Lambda を作って計測してみました。
先に結果だけ書くと、
- Rust: 55.8 k レコードを 23.4 秒で処理
=> 1 秒あたり約 2384.6 レコードを処理 - Python: 56.7 k レコードを 397 秒で処理
=> 1 秒あたり約 142.8 レコードを処理 - Golang: 52.4k レコードを 13.4 秒で処理
=> 1 秒あたり約 3910.4 レコードを処理
表にするとこんな感じ。
Runtime | record/s |
---|---|
Python | 142.8 |
Rust | 2348.6 |
Golang | 3910.0 |
すごい。
仮説
当初立てていた仮説では、Rust とそんなに差はでないかな、と考えていました。
というのも、風の噂でGolang の正規表現は遅い場合がある
と聞いていたので、ネイティブで動くにしてもそのあたりで差が詰まってトントンという見込みでした。
実装
実装が上です。これまで同様、 Apache のログを JSON に変換する Lambda です。
AWS CodeStar で作った Web サービス用のものをゴリゴリ書き換えました。
Golang は初めて書きましたが、あんまり違和感なくいけました。
C 書いてましたし、error まわりも Rust の Result
と似たような意図を感じたので、そのまま書きました。
ただ、ただ日付には戸惑いました… わかるか!!
そんな僕を助けてくれた記事です。3年以上前…
Goのtimeパッケージのリファレンスタイム(2006年1月2日)は何の日? - Qiita
測定(おさらい)
Lambda の環境
- Runtime: Go 1.x
- メモリ: 128 MB
- 実行時間上限: 5分
Runtime は Go、残り2つは前回と同様です。
指標
ExecuteProcessing.Duration
: 変換 Lambda の実行時間SucceedProcessing.Records
: 変換 Lambda が正常に変換できたレコード数
こちらも前回同様です。
正常に変換できたレコード数を、どれだけの実行時間でさばいたかで測ります。
結果
52.4k レコードを 13.4 秒で処理してます。
ので結果として、1 秒あたり約 3910.4 レコードを処理 した、ってことがわかりました。めっちゃ速いな!
予備計測
上の実装では json-iterator
という速い package を使っていたためだきっとそうだ、ということで、JSON へのエンコードを標準の encoding/json
でも行いました。
その結果がこちら。
56.2k レコードを 15.8 秒で処理してます。
1 秒あたり約 3557.0 レコードを処理 てことでやっぱり速い!
まとめ
仮説は外れ、Golang は速かったです。
くやしい… Rust の正式対応こい!