[Blog Reading] Neflix Platform Engineering — we’re just getting started

　エンジニアリングの世界に終わりはない、常に新しいチャレンジが待っているという内容のブログ。

Neflix Platform Engineering — we’re just getting started

• 2, 3年ほど前にKafkaを使うようになった
• 最近はApache Flinkをストリーム処理に使うようになった
• AMI/VMからcontainer technologyに移行
• 数100のマイクロサービスが動いている
• Chaos Monkey（NetflixのOSS）からChaos Engineeringという規範が生まれた

[Blog Reading] Building a Real-Time Streaming ETL Pipeline in 20 Minutes

Confluentのブログを読んだ。
最近KafkaまわりのOSSにプルリク送ってみたりして、Confluent熱が個人的に高まっている。

Building a Real-Time Streaming ETL Pipeline in 20 Minutes

以下、感想・考えてみたことなど。

• Kafkaってなんなのか一言で説明するのは難しいが、このブログにあるように「分散ストリーミング基盤」というのが良さそう。
• Streaming ETLという概念がやばい。Batch ETLでやっているような集計処理をバックグラウンドでリアルタイムに実行できるやつ。
• ブログの下の方で引用されているが、簡単なサンプルアプリがあるので見てみるとイメージがわく
• 「複数のアプリが同じデータソースを参照していると密結合になる」みたいな話を聞いたことがあったが、その感覚が分かった。例えば、アプリAがデータ生成するテーブルAがあったとして、これをみたい他のアプリB, C, D, ..があったとする。アプリB, C, D, ..からテーブルAを直接参照すると以下のような問題が起こる。
• 参照者が増えてくると、テーブルAのアクセス負荷が増える（Producer:Consumer = 1:N）
• アプリAが自由にテーブルAのスキーマを変更できない（テーブルAの変更は参照側のアプリに影響を与える）
• 参照で集計処理をしたい場合は、日次バッチになってしまう（負荷が高いのでリアルタイムで集計できない）

VagrantとAnsibleでKafka環境をつくる（８）

　Vagrant、Ansibleで作ったKafka環境で動作確認をしてみる。
以下の操作はすべてプロジェクトのrootディレクトリから実行するものとします。

まずvagrantで仮想マシンを起動する。

$ cd vagrant
$ vagrant up

続いてansibleでplaybookを適用する。

$ cd ansible
$ ansible-playbook site.yml 

まずはrest proxyと疎通確認を行う。

$ curl 192.168.33.11:8082/topics
["_schemas"]

rest proxy経由でKafkaにデータを送ってみる。

$ curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
 --data '{"records":[{"value":{"name": "testUser"}}]}' \
      "192.168.33.11:8082/topics/jsontest"
{"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

consumerを作成し、先ほど作成したjsontestトピックをサブスクライブする。

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" -H "Accept: application/vnd.kafka.v2+json" \
--data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}' \
http://192.168.33.11:8082/consumers/my_json_consumer
{"instance_id":"my_consumer_instance","base_uri":"http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance"}

$ curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["jsontest"]}' \
http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance/subscription

もう1つメッセージを送ってみる。

$ curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
 --data '{"records":[{"value":{"name": "testUserXXX"}}]}' \
      "192.168.33.11:8082/topics/jsontest"
{"offsets":[{"partition":0,"offset":1,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

メッセージをコンシュームする。

$ curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \       
http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance/records                               
[{"key":null,"value":{"name":"testUser"},"partition":0,"offset":0,"topic":"jsontest"},{"key":null,"value":{"name":"testUserXXX"},"partition":0,"offset":1,"topic":"jsontest"}]

データの送受信を確認できたので、テスト用のコンシューマーを削除する。

$ curl -X DELETE -H "Accept: application/vnd.kafka.v2+json" \
      http://192.168.33.11:8082/consumers/my_json_consumer/instances/my_consumer_instance

データフォーマットはどうあるべきか

　データパイプラインを作るときにデータのフォーマットはどうあるべきかについて、技術系のブログをいくつか読みつつ考えてみた。

Part of the problem is performance — text formats have to be parsed every time they are processed. Data is typically written once but processed many times; text formats add a significant overhead to every data query or analysis.

But the worst problem by far is the fact that with CSV and JSON data, the data has a schema, but the schema isn’t stored with the data. For example, CSV files have columns, and those columns have meaning. They represent IDs, names, phone numbers, etc. Each of these columns also has a data type: they can represent integers, strings, or dates. There are also some constraints involved — you can dictate that some of those columns contain unique values or that others will never contain nulls. All this information exists in the head of the people managing the data, but it doesn’t exist in the data itself.

The people who work with the data don’t just know about the schema; they need to use this knowledge when processing and analyzing the data. So the schema we never admitted to having is now coded in Python and Pig, Java and R, and every other application or script written to access the data.

引用元: The problem of managing schemas
https://www.oreilly.com/ideas/the-problem-of-managing-schemas

•  データフォーマットはデータ分析のパフォーマンスに影響をあたえる（パースするためのオーバーヘッド）。
• データはwriteの一度だけだが、readは複数回であるということを意識するのが重要。多少面倒でもwriteするときによいフォーマットでデータを出力しておけば、readするときに多くの人がhappyになるという 意味。
• CSV、JSONにスキーマ情報を定義することはできるが、データそのものの中にスキーマ情報を格納できない。
• データの中に格納されなかったスキーマ情報に関する処理はそのデータを利用する各アプリケーション内のコードに書かれることになる。つまり同じロジックがいろいろな場所にいろいろな言語で書かれることになる。

One of the key design decisions of Yelp’s Data Pipeline is schematizing all data. In other words, the pipeline forces all the data flowing through it to conform to predefined schemas, instead of existing in a free form. Standardizing the data format is crucial to the Yelp Data Pipeline because we want the data consumers to have an expectation about what type of data they are getting, as well as being able to avoid immediate impact when the upstream data producers decide to change the data they publish. Having a uniform schema representation also gives the Data Pipeline a really easy way to integrate and support various systems that use different data formats.

引用元：More Than Just a Schema Store
https://engineeringblog.yelp.com/2016/08/more-than-just-a-schema-store.html 

• データパイプライン上を流れるデータにあらかじめ定義されたスキーマ制約を課しておくことはとても重要
• フォーマットを標準化しておくことでconsumerはどのような形式のデータを取得するのか予測できる
• 統一的なスキーマ表現は異なる形式のデータをもつ多様なシステムを統合する際にも重要 

A common pain point that every data pipeline must address is how to deal with upstream schema changes. Often times such an event requires a lot of communication and coordination between upstream data producers and downstream data consumers. Yelp is not immune to this problem. We have batch jobs and systems that ingest data coming from other batch jobs and systems. It has become a painful problem every time the upstream data changes because it may cause the downstream batch jobs or systems to crash or necessitate a backfill. The entire process is pretty labor-intensive.

引用元：More Than Just a Schema Store
https://engineeringblog.yelp.com/2016/08/more-than-just-a-schema-store.html 

• アップストリームでのスキーマ変更にどのように対処するかはデータパイプラインの共通的な悩み
• あるバッチの出力が別のバッチの入力となっている場合、アップストリーム側のデータが変わるとダウンストリーム側のバッチに影響を及ぼす
• スキーマ変更が生じると、producer、consumer間のコミュニケーション、調整に多大な工数がかかる 

ここまでがデータフォーマットを統一することの重要性、データ内にスキーマ情報をもつことの重要性に関する話。ここからは、どのフォーマットがいいのか、スキーマの進化に強いフォーマットは無いのかという話。YelpではAvroを使っているらしい。

Apache Avro, a data serialization system, has some really nice properties, and is ultimately what we selected. Avro is a space-efficient binary serialization format that integrates nicely with dynamic languages like Python, without requiring code generation. The killer feature of Avro, for our system, is that it supports schema evolution. That means that a reader application and a writer application can use different schema versions to consume and produce data, as long as the two are compatible. This decouples consumers and producers nicely - producers can iterate on their data format, without requiring changes in consumer applications.

引用元：Billions of Messages a Day - Yelp's Real-time Data Pipeline
https://engineeringblog.yelp.com/2016/07/billions-of-messages-a-day-yelps-real-time-data-pipeline.html

• Avroは容量効率がとても優れているバイナリシリアライゼーションフォーマット
• スキーマの進化をサポート
• スキーマに互換性があれば、producerが書くデータ、consumerが読むデータが異なるバージョンであってもよい
• つまり、consumerとproducerの結合度が弱まる
• producer側はconsumer側の影響を意識せずにデータフォーマットを継続的に改良することができる

まとめ

• データパイプラインを流れるデータはフリー形式ではなく、なんらかのフォーマットに統一するべき
• フォーマット形式は
• データ圧縮率が高いものがよい
• データ内にスキーマ情報を持てるものがよい
• スキーマの進化をサポートしているものがよい
• 適切なフォーマット形式を選ぶことで以下のメリットがえられる
• データ分析処理のパフォーマンスが向上
• producer/consumer間のコミュニケーションコストが減る
• アップストリーム側の変更がダウンストリーム側に及びにくくなる
• スキーマ情報に関する処理コード（型変換、デフォルト値の設定、単純なパース処理）の記述量が減る

Message QueueとPublish-Subscribeの違い

Messaging Systemには大きく二つの種類がある。
一つがMessage Queueで、もう一つがPublish-Subscribe。

それぞれの特徴、長所、短所を簡単にまとめておく。

	Message Queue	Publish-Subscribe
特徴	一つのレコードは一つのConsumerが消費する。	一つのレコードはすべてのConsumerに配信される。
長所	データ処理を複数のConsumerインスタンスで並列化できる。	一つのデータを複数のConsumerで利用することができる。
短所	一つのConsumerがデータを読むと、そのデータは消える。	データ処理を並列化できない。