2020-01-31

Schema Registry について書いていくその1: Confluent Schema Registry

分散アプリケーション間のメッセージングやログ収集基盤において、しばしばスキーマの扱いは便利である反面頭を悩ませる種になります。

スキーマを厳密に定義して、 Protocol Buffers や Avro などのシリアライゼーションフォーマットを用いることで、メッセージのサイズの減少やメッセージの扱いの安全性を実行時より早く検出できる機会が増えます。しかしスキーマの互換性に気をつけたり、メッセージをシリアライズ・デシリアライズするアプリケーション間でスキーマ情報をどう共有するかの課題も発生します。この課題は、昨年発売された "データ指向アプリケーションデザイン" でも "4.1.4.3 そもそもライターのスキーマとは何なのか？" などで触れられています。

世の中にはこの課題を解決するための "スキーマレジストリ" と呼ばれるソフトウェア、サービスがいくつか存在します。残念ながら日本語の資料があまり無いようにも感じますし、筆者の気が済むか飽きるまで様々なスキーマレジストリを探索してみようかと思います。第一回は Confluent Schema Registry について調べてみます。（残念ながら筆者は Confluent Schema Registry を実務で触れた経験がなく、なにか内容に誤りがある可能性があります）

Confluent Schema Registry とは

Confluent Schema Registry は、マネージドな Apache Kafka を提供する Confluent 社が開発したスキーマレジストリです。

docs.confluent.io

以下のような機能を持っています。

Apache Kafka に組み込む想定で、 Apache Avro のスキーマを管理
Subject 名というスキーマレジストリ用のスキーマ解決の為のキー情報を持つ
Kafka の専用 topic にスキーマ情報を格納する
互換性チェックの機構を持つ
REST API を提供する
管理用の WebUI でスキーマ管理を行える

Apache Avro については以前本ブログで触れましたし世の中にドキュメントが多数存在するのでここでは別段触れません。

スキーマは Subject という識別に用いる名前とバージョン番号で管理されます。 Subject をどう決めるかはいくつかのオプションがあり、デフォルトでは topic 名になっています。 topic 名を用いる場合はある topic に流れるレコードのスキーマは一つに限定され、複数の異なるスキーマのレコードを流すことは出来ません。これを行いたい場合は他にレコード名もしくは topic 名とレコード名の組み合わせを Subject として用いることになります。

スキーマ情報の格納先がそれもまた Kafka であるのはちょっとユニークな点と言えそうです。デフォルトではシングルパーティションの _schemas topic にスキーマ情報を格納するようです。

docs.confluent.io

スキーマレジストリという機構の悩ましい点として、それ自体がメッセージバスと同等くらいの可用性を求められることが挙げられます。メッセージバスが生きていてもスキーマレジストリが死んでいるなら、シリアライズする時に用いたスキーマを何処にも格納できず、デシリアライズする側のシステムに伝える術も失うためです。 Confluent Schema Registry ではスキーマレジストリの REST API を提供する HTTP サーバの可用性の懸念は残りますが、ストレージは Kafka それ自体と運命共同体になるため考える課題が一つ減りそうです。

Confluent Schema Registry は更にスキーマの更新問題にも解決策をもらたします。アプリケーションを稼働させていくうちにスキーマを更新したくなるシーンが登場するのは自然なことでしょう。この際に Confluent Schema Registry は Kafka の Producer が作成した Avro のスキーマは解決してくれますが、 Consumer 側にも配慮する必要があります。例えば Consumer が必要としているフィールドがある日 Producer が送信するレコードから無断で削除されたなら、 Consumer 側で何らかの問題が出るかもしれません。この問題に配慮するため、 Confluent Schema Registry ではスキーマの互換性チェックの機構を設けています。互換性には前方互換性、後方互換性、完全互換性の 3 タイプが設けられています。

docs.confluent.io

前方互換性は Producer が Consumer より先にスキーマを更新するパターンで求められる互換性です。この場合フィールドの追加もしくは optional 、すなわちデフォルト値を持つフィールドの削除が許されています。これらの変更が行われても Consumer は、自分が知らない追加フィールドは無視する、削除されたフィールドはデフォルト値で埋めることができます。

f:id:syu_cream:20200131004548p:plain

後方互換性は Consumer が Producer より先にスキーマを更新するパターンで求められる互換性です。この場合は前述の逆で、 optional なフィールドの追加とフィールドの削除が許されています。この互換性タイプが Confluent Schema Registry のデフォルトの互換性モードであり、現実世界で気を使われることが多いものであると思います。

f:id:syu_cream:20200131004611p:plain

完全互換性は Producer と Consumer どちらが先にスキーマを更新しても良いパターンで求められる互換性です。この場合は制約が最も弱いようなスキーマ更新、つまり optional なフィールドの追加・削除が許されるようになります。この互換性タイプは制約が弱いため扱いにも困るケースが出るかも知れませんが、 Confluent のプラットフォーム以外のミドルウェアやキャッシュ機構が存在したり処理するレコードの順序が保証されなかったりするケースに、互換性の問題で Consumer 側でエラーが発生するケースを低減できそうです。

f:id:syu_cream:20200131004629p:plain

Confluent Schema Registry はスキーマを更新しようとする際に、求められるタイプに従って互換性をチェックする機能を提供します。

REST API は実際に Confluent Schema Registry に対応した SerDe でスキーマ操作を行う際に使われる API にもなります。手動で REST API を叩いてスキーマ登録を行うなどをしても良いでしょう。この API の仕様は以下にまとまっています。

docs.confluent.io

管理用の WebUI については実際に触れてみた方が話が早そうです。次に触れてみましょう。

Confluent Schema Registry に実際に触れてみる

クイックスタート

公式で Docker イメージと docker-compose yml ファイルが提供されており、とりあえず動かす敷居が低めになっています。

docs.confluent.io

このドキュメントに沿って Docker コンテナを動かした後にブラウザ上で http://localhost:9021/ にアクセスしてみます。上手く動いていれば以下のような管理画面を閲覧できるはずです。

f:id:syu_cream:20200129004253p:plain

Avro のスキーマを登録してみる

ここでは実験用に hello topic を作成して、この topic に流すレコードの value のスキーマを設定してみようと思います。

ここではバージョン 1 のスキーマとして以下を与えてみます。

{
  "name": "Hello",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "long"
    },
    {
      "name": "name",
      "type": "string"
    }
  ]
}

登録したスキーマを WebUI 上で確認できます。

f:id:syu_cream:20200131001440p:plain

一度スキーマを登録すると互換性モードを選べるようになります。ここでは一つ前のバージョンのスキーマと後方互換性を保つモードである Backward に設定しておきます。

f:id:syu_cream:20200131001705p:plain

Avro のスキーマを更新してみる; 互換性がある場合

先程登録したスキーマを更新してみようと思います。ここでは年齢を指定する想定で {"name": "age", "type": ["null", "long"], "default": null} という nullable なフィールドを追加してみます。

{
  "name": "Hello",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "long"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "age",
      "type": ["null", "long"],
      "default": null
    }
  ]
}

このスキーマ変更は optional なフィールド追加に相当して後方互換性チェックをパスすることができます。無事 WebUI からも新しいバージョンのスキーマを閲覧することができています。 diff 表示も出来ていい感じです。

f:id:syu_cream:20200131002225p:plain

Avro のスキーマを更新してみる; 互換性がない場合

ここから後方互換性が無い変更をしてみます。よく考えたらこのレコードをアプリケーション上で作成した時刻の timestamp 値を保持したくなってきました。ということで optional でないフィールド {"name": "created_at", "type": "long", "logicalType": "timestamp-millis"} を追加してみます。

{
  "name": "Hello",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "long"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "age",
      "type": ["null", "long"],
      "default": null
    },
    {
      "name": "created_at",
      "type": "long",
      "logicalType": "timestamp-millis"
    }
  ]
}

しかしながら先述の通り後方互換性を保つには optional なフィールド追加は許されません。この場合 Confluent Schema Registry が互換性が保てない旨のメッセージを出してスキーマの更新が拒否されます。

f:id:syu_cream:20200131002951p:plain

このように Confluent Schema Registry では、スキーマの管理と共に、求められる互換性のタイプに従ってそれを維持するようチェックをかけてくれます。本当はスキーマ管理を .avsc ファイルなどで保存して CI で互換性チェックし、パスすればスキーマ更新するなど人間による直接の更新を避けるのが望ましいかもですが、この WebUI と互換性チェック機構で人間による直接管理もある程度運用に耐えるかもしれません。

余談; `_schemas` topic

WebUI 上では _schemas topic の様子も確認することができます。 log.cleanup.policy が delete ではなく compact が設定されており、これにより時間経過等によるレコードの削除を抑制しています。

f:id:syu_cream:20200131010901p:plain

Confluent Schema Registry の実装を覗いてみる

Confluent Schema Registry のコードはGitHub で公開されています。折角なのでどんな実装になっているか少し覗いてみます。ちなみにここでは網羅性などは全く考慮していません。筆者が気になるポイントを勝手にかいつまんでいます。

github.com

ここで紹介する内容は v5.4.0 時点のものとなります。

avro-serializer: Schema Registry と通信した結果でシリアライズ・デシリアライズを行う

この module に存在するのが、実際に producer, consumer を実装する際に properties に指定する serializer, deserializer になります。

シリアライズのコアの実装が AbstractKafkaAvroSerializer の serializeImpl() に存在します。自動登録する設定になっているなら Avro スキーマをスキーマレジストリに登録し、その後 Kafka の topic に送信するメッセージをシリアライズします。この時のバイナリレイアウトなのですが、この Serializer 実装特有のフォーマットになっており Avro のレコードの手前に 8 バイトのマジックナンバーとスキーマの ID が埋め込まれています。 Avro のレコードのバイナリはおなじみ SpecificDatumWriter, ReflectDatumWriter, あるいは GenericDatumWriter で書き出します。

docs.confluent.io

これに対応してデシリアライズ側の実装は AbstractKafkaAvroDeserializer クラスに多くのロジックが持たれています。まずは deserialize() ですがリーダ側のスキーマが渡せるようになっており、デシリアライズの結果をライター側とは異なる（しかし互換性はある）型に展開できるようになっています。ライター側のスキーマは勿論スキーマレジストリ経由で解決できます。この辺りの実装は AbstractKafkaAvroDeserializer の内部クラスの DeserializationContext クラスで行われます。このクラスに渡された、デシリアライズ対象のバイナリは、先頭 8 バイトにマジックナンバーとスキーマ ID が存在するはすです。まずマジックナンバーの存在チェックを行った後にスキーマ ID を取り出し、この ID をスキーマレジストリに問い合わせて Avro のスキーマ情報を取得します。その後はやはりおなじみ SpecificDatumReader, ReflectDatumReader, GenericDatumReader で Java のオブジェクトとして読み出します。

これで概ね Avro のレコードのシリアライズとデシリアライズの流れが掴めますが、スキーマの登録と取得がどのように成されているのでしょうか。先述の Serializer, Deserializer の共通の親クラスとして AbstractKafkaAvroSerDe があり、このクラスがデフォルトでは CachedSchemaRegistryClient というスキーマレジストリに対するクライアントを保持しています。このクライアントの実装がどうなっているのか、次はスキーマレジストリとの通信を行っている client module も覗いてみます。

client: Schema Registry のクライアント

SchemaRegistryClient インタフェースを実装したクラスが実際に Schema Registry と通信しています。先述の CachedSchemaRegistryClient はその 1 実装であり、スキーマや ID, バージョン情報のキャッシュと RestService というクラスのメンバを持ちます。そして RestService が実際に HTTP リクエストをスキーマレジストリの REST API に送信する、例えばスキーマの登録時に /subjects/:subject/versions に POST リクエストを送る、というような格好になります。

core: Schema Registry のコア実装

もうひとつ、スキーマレジストリの REST API サーバを提供する core モジュールも気になる存在です。 REST API の実装やスキーマ情報などの永続化をどのように行っているのでしょうか。

スキーマレジストリの REST API サーバは Jetty + Jersey で実装されているようです。 API の各エンドポイントは JAX-RS の Resource として実装されており、例えば /subjects/{subject}/versions であれば SubjectVersionsResource が相当する、などの構成になっています。

スキーマなどの情報を永続化するストレージレイヤは KafkaStore クラスで提供されています。スキーマの登録のため put() が呼ばれると、スキーマ情報を ProducerRecord に包んで _schemas topic に送ります。他方、スキーマの情報取得はというと KafkaStore 自体では直接は _schemas topic を consume せず、別スレッドで動作してローカルスキーマキャッシュを更新する KafkaStoreReaderThread が更新した InMemoryCache から取得する格好になります。この動作についてはドキュメントでも言及されています。

docs.confluent.io

おわりに

Confluent Schema Registry について筆者の思うまま気になるところをざっくり紹介しました。いかがでしたでしょうか。

次回は Apache Pulsar の持つ Schema Registry 機能について調べて行こうかなと思います。なんとこちらのスキーマレジストリは Avro は勿論 Protocol Buffers もサポートしているようです！ Protocol Buffers は各言語向けの自動生成されたライブラリを使う、 Avro で言う SpecificRecord のサブクラスに近い性質を持つ表現で扱われることが多いと思われますが、スキーマレジストリをどのように提供しているのでしょうか。

pulsar.apache.org

余談ですが本記事執筆中に Confluent Schema Registry でも Protocol Buffers をサポートする気配のある pull request を発見してしまいました。近い将来 Avro 以外でシリアライズしたレコードが色んなミドルウェアでらくらくに扱える日が来るかも知れませんね。

github.com

2019-10-15

Apache Avro について知っていることを書いていく　その2

Apache Avro について書き下していく記事その 2 です。本記事では Avro で表現されるデータのプログラミング言語上の表現、特に Java を想定して SpecificData と GenericData について触れていきます。

Avro のデータ型とコード生成

前回の記事で触れた通り、 Avro では様々なリッチなデータ型を表現できます。多くの場合一番外側のスコープでは record type を使って構造体を定義することになるかなと思います。 Avro スキーマは以下のように事前に JSON 文字列を埋め込むか .avsc ファイルで保存しておくのもアリですが、実行時に動的生成することもできます。

{
  "name": "Hello",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "long"
    },
    {
      "name": "name",
      "type": "string"
    }
  ]
}

ここで定義したデータ型を実際にどう扱いましょうか？例えば Protocol Buffers ではちょうど Avro で .avsc ファイルに記述するように、以下のようにおおむね .proto ファイルにメッセージ型を定義していくことになると思います。

syntax = "proto3";

package com.syucream.example;

message Hello {
  uint64 id = 1;
  string name = 2;
}

その後典型的には protoc と各言語に対応する protoc plugin を用いることで、定義したメッセージ型に従ったコードを生成することになります。

$ protoc --java_out=./ hello.proto

// Generated by the protocol buffer compiler.  DO NOT EDIT!
// source: hello.proto

package com.syucream.example;

public final class HelloOuterClass {
  private HelloOuterClass() {}
  public static void registerAllExtensions(
      com.google.protobuf.ExtensionRegistryLite registry) {
  }

  public static void registerAllExtensions(
      com.google.protobuf.ExtensionRegistry registry) {
    registerAllExtensions(
        (com.google.protobuf.ExtensionRegistryLite) registry);
  }
  public interface HelloOrBuilder extends
      // @@protoc_insertion_point(interface_extends:com.syucream.example.Hello)
      com.google.protobuf.MessageOrBuilder {

    /**
     * <code>uint64 id = 1;</code>
     */
    long getId();

    /**
     * <code>string name = 2;</code>
     */
    java.lang.String getName();
   ...

つまりアプリケーションに組み込んで用いる際に事前にコード生成を求められます。

Avro ではコード生成はオプションです。事前に生成したコードを用いることも、そうでない選択肢を取ることもできます。

基本的なデータ型表現セット GenericData

org.apache.avro.generic.GenericData は Avro のデータ型の Java での表現の基礎となるクラスなどが詰まったユーティリティです。 GenericData 以下に具体的な型に対応したクラスが存在します。例えば record なら org.apache.avro.generic.GenericData.Record といった具合です。以下では特にこの GenericData.Record を掘り下げましょう。

まずこのクラスは org.apache.avro.generic.GenericRecord interface を実装しているので、まぁ名前の通りなのですがつまり値を get したり put したりできます。一見あまり情報量が無さそうなこの interface は、結構色々な Java 製ミドルウェアで頻出します。またこの get() の戻り値は Object 型であり、実際どんな型なのか意識するのは get() を呼び出すコードを書く人の責務になります。

GenericData.Record オブジェクトを作るには、 Schema オブジェクトを渡して new します。 Schema オブジェクトは Schema.Parser を使って事前定義した JSON 文字列をパースして得るのもよし、動的に Schema.createRecord() などから生成してもよし、です。

この GenericData.Record ですが、フィールドの具体的な値は Object[] 型のメンバとして持ち、 get/put などをする時は Schema オブジェクトの情報からフィールドのインデックスを得てアクセスします。これはちょうど、 Avro の仕様としてシリアライズした後のバイナリのレイアウトがどうなっているかがスキーマによって決定づけられる（他にレイアウト情報を持たずスキーマを信じる）のに対応しそうです。

GenericData を読み書きする GenericDatumReader/GenericDatumWriter

org.apache.avro.generic.GenericDatumReader を用いることでシリアライズされた状態のバイナリから GenericData 以下の Java のオブジェクトを取り出すことができます。ジェネリクスのパラメータは、例えば record 型の値を参照するだけでいいなら前述の GenericRecord を指定するので良いでしょう。逆に Java のオブジェクトをシリアライズしたい際は org.apache.avro.generic.GenericDatumWriter を使うことができます。より具体的な使い方は公式のテストコードを見ると参考になります。

github.com

Avro のシリアライズ方法とバイナリレイアウトを決めているのはスキーマです。というわけでもちろん、GenericDatumReader / GenericDatumWriter を用いる際にはスキーマ情報が求められます。

さて実は GenericDatumReader には二種類のスキーマ、引数名でいうと reader と writer を与えることができます。名前の通り writer は GenericDatumWriter を使ってシリアライズする側で用いられた側のスキーマ、 reader は GenericDatumReader で読み出す側のスキーマです。具体的で実用しそうなシナリオとしては、 writer が古いスキーマでシリアライズする可能性のある環境下で reader としては新しいスキーマ（古いスキーマからの後方互換性がある）に従う型のオブジェクトとして扱いたい際に二種類のスキーマを与える場合がありそうです。

より安全で高速な表現セット SpecificData

さてこの GenericData 関連のツールですが、値の取り回しは Object 型ですることになるので逐一キャストしなければならず、安全な実装をできるかやキャストするコストが発生する点が気になります。悪い言い方をすれば、シリアライズ・デシリアライズの操作を含めてスキーマに違反していないことをは保証された値でしかなく、 Java のコード上は例えば record 型なら単なる Map<String, Object> 型オブジェクトと言えそうです。

幸いなことに Avro では GenericData を使うのとは別に SpecificData サブクラスを使う道も存在します。 org.apache.avro.specific.SpecificData は GenericData のサブクラスです。と言ってもこのクラス自体にはさほど機能は追加されていません。重要なのはこれを継承した、自動生成された schema specific なクラスです。

SpecificData なクラスのコード生成

最初に述べた通り Avro には事前に .java ファイルのコードを生成しておく道もあります。コード生成は avro-tools を通して行えます。試しにこの記事の冒頭に出てきた .avsc からコード生成してみます。

$ java -jar ~/tools/avro-tools-1.8.2.jar compile schema hello.avsc ./
Input files to compile:
  hello.avsc
log4j:WARN No appenders could be found for logger (AvroVelocityLogChute).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

得られる Java のコードは以下のようになります（一部抜粋）

/**
 * Autogenerated by Avro
 *
 * DO NOT EDIT DIRECTLY
 */

import org.apache.avro.specific.SpecificData;
import org.apache.avro.message.BinaryMessageEncoder;
import org.apache.avro.message.BinaryMessageDecoder;
import org.apache.avro.message.SchemaStore;

@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class Hello extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
  private static final long serialVersionUID = -8266440640401408575L;
  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Hello\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"},{\"name\":\"name\",\"type\":\"string\"}]}");
  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }

  ...

  @Deprecated public long id;
  @Deprecated public java.lang.CharSequence name;

  /**
   * Default constructor.  Note that this does not initialize fields
   * to their default values from the schema.  If that is desired then
   * one should use <code>newBuilder()</code>.
   */
  public Hello() {}

  ...

  /**
   * Gets the value of the 'id' field.
   * @return The value of the 'id' field.
   */
  public java.lang.Long getId() {
    return id;
  }

  /**
   * Sets the value of the 'id' field.
   * @param value the value to set.
   */
  public void setId(java.lang.Long value) {
    this.id = value;
  }

  /**
   * Gets the value of the 'name' field.
   * @return The value of the 'name' field.
   */
  public java.lang.CharSequence getName() {
    return name;
  }

  /**
   * Sets the value of the 'name' field.
   * @param value the value to set.
   */
  public void setName(java.lang.CharSequence value) {
    this.name = value;
  }

  ...

嬉しいことにフィールドの型に対応した getter/setter が手に入ります！ GenericRecord interface のメソッドを使うよりこちらの方が安全で若干高速になるはずです。

見てきた通り SpecificData 系サブクラスはコード生成する都合コンパイル時にスキーマが決まっている必要があります。またスキーマ定義が頻繁に変更され、その度にコード生成し直しまくるのもやや煩雑かと思われます。あまり変わらない・変えたくないデータ型を定義する時は SpecificData からなるクラスを生成、動的に・頻繁に変わる可能性があるデータ型は GenericData として扱うのが良いかもしれません。

Java 意外の言語ではどうなる？

Java の例を見てきた通り Specific なテータ表現は Generic なもののサブセットとなっています。そんな訳で多くのプログラミング言語では Generic なデータ表現はサポートされています。

が、 Specific によってはサポート状況がまちまちなようです。筆者が認識した限りでは、 C++ で experimental でサポート? されていたり、 Go では gogen-avro を作っている人がいます。

飽きてきたので

本記事はこのあたりで終わります。あとは筆者がよくハマる、複雑なスキーマを定義してみる話やシリアライズされた後のバイナリのレイアウトの話とか別記事で書くかもしれないです。

2019-10-09

Apache Avro について知っていることを書いていく　その1

Apache Avro になにかと縁があり、かつ普及しているテクノロジーの割に日本語の情報がそんなにない（個人の意見です、意外とあるかも）のでつらつら書いてみます。整理はされておらずシーケンシャルに要素を並べています。実装についてとくに言及がされていなければ、 Java の 1.8.2 について触れているものとします。

Apache Avro はデータフォーマットとエコシステムのひとつ

Apache Avro は Apache トップレベルプロジェクトのひとつでファイルフォーマットとその周辺エコシステムです。比較される技術として Protocol Buffers や Message Pack が挙げられると思います。 Apache Avro はそれらの中でも主に Hadoop エコシステムなどビッグデータ絡みの文脈でよく登場するように思えます。

以下のディレクトリ構成の通り、公式でいくつかのプログラミング言語をサポートしているようです。

github.com

筆者はほとんど Java のパッケージしか使わず正確な比較も行ってないのですが、おそらく Java 向けの機能がサポート厚めだと思われます。 avro-tools が Java で書かれてたり avro-protobuf という Protocol Buffers との変換ライブラリが提供されていたりするので。また非公式ですが以下のような他言語向けライブラリもあったりします。

github.com

Avro はスキーマの表現力が強い

Avro はスキーマに従いシリアライゼーションとでシリアライゼーションをするわけですが、このスキーマの表現力が割と強いです。雑に列挙すると以下の通りです。

record 型(構造体)のサポート
- エイリアスや doc 、 namespace の宣言もできる(オプション)
array 型(配列)のサポート
union 型(共用体)のサポート
- null 型との union を取り、デフォルト値を null にすることで nullable な型を表現できる
map 型のサポート
enum 型のサポート
fixed 型という固定長バイト配列のサポート（使ったこと無いのでよくわからん）
デフォルト値が設定できる
logical types を使うことで型に別の解釈をもたせることができる
- メジャーな使い方として long 型に timestamp-millis logical type をもたせるなど
- timestamp の制度は micros も選べる

これらにより、かなりリッチなスキーマを記述することができるはずです。

Avro のスキーマを記述するのがつらい

表現力が高いこともあるのですが、基本的にスキーマは JSON で記述することになる上か結構冗長でかつ人間にとって読みにくい内容になりがちです。公式リポジトリに簡単な例があるのですが、簡単なものでもある程度事前知識を求められるでしょう。さらに型をネストし始めたら地獄です。筆者は特殊な訓練を繰り返すことにより最近手でスキーマを書けるようになってきた気がしますが、このスキーマの記載を同僚に求めるのは酷なものです。

github.com

この問題はかなり認知されているのか、公式が IDL を提供しています。一見 Protocol Buffers を彷彿とさせる気がするシンタックスですね。

avro.apache.org

また、プログラムから動的にスキーマを記述、生成することもできます。 Java であれば Schema.createRecord() して List<Schema.Field> オブジェクトを埋めて・・・といった流れで実現できます。

Avro 自体の圧縮効率はそんな高くない？

int, long は zigzag encoding してくれますが、 bytes, string は length + バイナリなどかなりシンプルなシリアライズをします。結果としてシリアライズされたあとのバイナリのサイズは Protocol Buffers や Message Pack と比較してそんなに潰れないと思います。（実測した訳でなく、単にこれまでの経験則程度の話です）ただ Object Container Files フォーマットを取るとブロック単位に圧縮をかけてくれたりします。

Object Container Files フォーマットの便利さ

あなたがデータアナリストやデータサイエンティストなどのポジションで働いているなら、このフォーマットで Avro と対面することが多いかもしれません。 .avro 拡張子をよく取る、 Object Container Files フォーマットに従うレイアウトを取ったファイルです。

実体は well-known な Avro スキーマに従ってシリアライズされたファイルになります。そのスキーマの中にはレコードのスキーマとコーデック、フィンガープリント情報などが含まれます、ここのスキーマ情報により、ファイルを読むことで中のバイナリがどんなスキーマでシリアライズされたかわかるので動的にデシリアライザを生成して読める状態になってるわけですね！すぐれもの！

このファイルフォーマットは BigQuery や Redshift でもサポートされており、 Hadoop エコシステムにあまり関わらずとも Avro のファイルをオブジェクトストレージに書き出しておけば、あとで DHW に統合して安全便利にクエリを投げられるわけですね！すぐれもの！

avro-tools が開発運用に便利

あまり言及されてない気がしますが、 avro-tools という Java 製のツールが便利です。事前定義した Avro のスキーマを記載した.avsc ファイルから Specific クラスを自動生成するのはもちろん、 Object Container Files フォーマットのファイルから schema を読んで吐いてくれたりレコードを json に変換して出力してくれたりします。特にレコードを json で出力してくれると、 jq などと組み合わせてフィルタしたり内容確認できたりして便利ですね。

mvnrepository.com

飽きてきたので

本記事ではこれくらいにしておきます。気力があったり次回書きたいことがまとまってきたらまた書きます。

2019-07-24

「データとML周辺エンジニアリングを考える会」という勉強会の第二回を開催しました

TL;DR

2019/07/19(金)に、ヤフー株式会社様コワーキングスペースの LODGE において、「データとML周辺エンジニアリングを考える会」という勉強会の第二回目を開催しました。

data-engineering.connpass.com

データエンジニアリングとサイエンス、アナリティクスにざっくり被るような少し広く曖昧なドメインで開始した勉強会です。

今回は前回より少々多めで、 40 人強の参加者が参加してくれました。発表、 LT に合わせて懇親会も議論が弾み、主催のひとりとしては実現したかったことがある程度できたのではないかと考えています。

つらつら発表内容

せっかくなので覚えているうちに発表資料のリンク記載と超個人的な感想載せてみます。

15 分枠

GCPでStreamなデータパイプライン運用しはじめた by @shoe116

メルカリでのログ収集のためのパイプラインの構築の話です。マイクロサービスアーキテクチャへの移行やビジネス・組織のスケールに合わせた試行錯誤の跡が伺えます。というかわたしも業務で参加してるやつです。実際上記の試行錯誤をしています。

speakerdeck.com

行動ログ処理基盤の構築 by @hirosassa

サービスにおける行動ログの収集基盤の刷新の話です。現行システムが pull 型で基盤側がサービスの内情を知ってしまう問題を、 push 型のアーキテクチャにしたあたりが今後の投資になりうるおもしろい点なのではと思います。（基盤システムってどこまでサービスのことを知るか、責任分界をどこでするかしばしば悩ましくなりますよね）

speakerdeck.com

LT 枠

Google Cloud ML Engineに浸かってみる by @yudeayase

GCP の ML Engine の話です。 ML Engine, 便利そうではあるもののこの仕組みに特化してしまうのは良いのか？などと考えさせられました。

cloud.google.com

（資料アップロードなし？あとで調べる

PoC案件が多すぎてつらいので、パイプラインを使いまわすツールを入れた。 by @mori_kaz0429

繰り返し発生する PoC 案件で、似たようなクエリを投げたりすることが多い処理を共通化、再利用可能にする話です。最後の方には Apache Airflow などワークフローエンジンを今後使ってみたい話も。

（資料アップロードなし？あとで調べる

Cloud Composer & Cloud Dataflow によるバッチETLの再構築 by @yuzutas0

Cloud Composer (Apache Airflow のマネージドサービス) を使って壊れかけのデータ同期の仕組みを立て直す話です。データエンジニアとデータアナリスト、両方に対してヒアリングをかけつつ現状を鑑みて良いバランスのところを攻めるというマネジメントに近い側面もあれば、 Cloud Dataflow によるクレンジング処理に触れたりする話もあり盛りだくさんでした。

speakerdeck.com

DigdagでETL処理をする by @nakano_shota

今度は Digdag でワークフローを組んだ話です。 s3_touch というオペレーターを開発して、 s3_wait と組み合わせてプロジェクト間依存関係も対応できるようにしていて良い。あまり深く触れられなかったけどリトライ処理や冪等性担保大事ですよね。もっとお話聞いてみたい気がします。

speakerdeck.com

Comet.ml で AutoML ライブラリ開発（仮） by @Y_oHr_N

Comet.ml 、初めて知ったのですがかなり良さそう！ codecov でカバレッジを可視化するのと同じようにモデルの可視化をできるのは好感触！

speakerdeck.com

データ活用の際にハマってしまったログ・データスキーマ設計 by @yu-ya4

この手の苦労話、これこそこういう勉強会で話し合いたかったことな気がします。テーブルの日付が何の時間を表すか問題、スキーマ更新にどう立ち向かうか、 STRING 型フィールドに JSON 突っ込むのどうなんだ話、 null の扱いと結構あるある話な気がするが・・・。二番目の話題は個人的にも刺さるものがありました。簡易にアプリケーションからログを出力して、読み出す時に苦労するスキーマオンリードの戦略自体は間違ってはいないはずだし・・・。触れられていなかった別の点として、 BigQuery はカラムナでデータを持ってくれているはずなのですが、 JSON を突っ込んで読む時にパースするではパフォーマンスが落ちる（課金が増える？）かもしれないなと思いました。

speakerdeck.com

今後に向けて

幸いなことに参加者の方々からもそれなりに好評を得られたようだし、自分としても知見を得たい気持ちもあるため、主催メンバーで話し合いつつ第三回を企画していきたいと考えています。ジャストアイデアですが、初心者枠？というかこれからデータ基盤を作っていこうとする人たちが発表しやすい枠を設けるとかもあるとイベントの雰囲気変わるかなとか。その時が来たらまたアナウンスしますので、ご興味ある方々いらっしゃいましたらぜひぜひ！！

2018-12-21

技術書典5で配布した同人誌の原稿データを GitHub で公開しました

こちらで紹介した、僕が主催するサークルで技術書典5で配布した同人誌の原稿データを公開しました。

syucream.hatenablog.jp

リポジトリはこちらになります。

github.com

前回と同様、 Re:VIEW を使って記述しています。 epub / pdf ファイルが欲しい方はいい感じにビルドしてください。

もし多少でも気に入った方がいらっしゃれば、 Kindle 版を購入していただけると幸いです！僕のジュース代の足しくらいにはなります。

www.amazon.co.jp

2018-12-04

Kubernetes と CSI(Container Storage Interface) について

この記事は Kubernetes2 Advent Calendar 4 日目の記事です。

本記事は CSI(Container Storage Interface) と Kubernetes での CSI のサポートについて触れます。執筆に時間があまり割けなかった為、後でもう少し加筆する、あるいは別途続きの記事を書くかもしれません。

CSI(Container Storage Interface) とは

CSI(Container Storage Interface, 以下 CSI) は Kubernetes など複数の Container Orchestration (CO) から共通のプラグインを使って SP (Storage Provider) とやり取りするためのインタフェース仕様です。より具体的には、 CO からストレージを抽象化した Volume をアタッチ・デタッチしたりスナップショットを取ったりする機能のためのインタフェースを提供します。 CSI は Kubernetes とは独立したプロジェクトで進行しております。

github.com

また Kubernetes のプロジェクト以下で、 Kubernetes から SP が提供する CSI プラグインへ通信するための CSIDriver と関連実装を以下プロジェクトで進行しています。

github.com

Kubertenes としては CSI は 1.9 で Alpha 、 1.10 で Beta サポート対象になっていますまた Kubernetes での CSI 利用については他に既に試されている方もいるようです。こちらの記事は詳細に踏み込んだ構成説明などもされており、オススメできる内容になっています。

qiita.com

CSI の仕様

CSI のインタフェース仕様は Protocol Buffer で定義されており、以下リポジトリで管理されています。またこのリポジトリ内の lib/go/ ディレクトリには .proto ファイルの定義に従ったライブラリの Go 実装が配置されています。

github.com

CSI のプラグインは Controller plugin, Node plugin の二種類が想定されております。また CO とこれら CSI プラグイン間は gRPC で通信することが想定されています。

CSI のサービス

CSI のサービスとして以下三種類が想定されています。構成によっては実装しなくても良いサービスやメソッドがありますが、少なくとも Node Service の NodePublishVolume, NodeUnpublishVolume は実装する必要があります。（でないと Node から Volume を参照できないはず）

Identity Service
- CO から Plugin のケーパビリティ (後述) やヘルスチェック、メタデータ参照を可能にするサービス
Controller Service
- Node をまたいで Volume の構成を管理するためのサービス
- Volume のスナップショット操作もサポートする
Node Service
- 各 Node から Volume の操作を可能にするためのサービス

CSI における Volume の状態遷移

CSI で Node から Volume を利用されるまでに、 Volume は基本的に以下のような状態遷移をしていきます。幾つかの状態は、プラグインが示す後述の CSI ケーパビリティによって存在しなかったりします。

   CreateVolume +------------+ DeleteVolume
 +------------->|  CREATED   +--------------+
 |              +---+----+---+              |
 |       Controller |    | Controller       v
+++         Publish |    | Unpublish       +++
|X|          Volume |    | Volume          | |
+-+             +---v----+---+             +-+
                | NODE_READY |
                +---+----^---+
               Node |    | Node
              Stage |    | Unstage
             Volume |    | Volume
                +---v----+---+
                |  VOL_READY |
                +------------+
               Node |    | Node
            Publish |    | Unpublish
             Volume |    | Volume
                +---v----+---+
                | PUBLISHED  |
                +------------+

ref. https://github.com/container-storage-interface/spec/blob/master/spec.md#volume-lifecycle

CSI のケーパビリティ

CSI プラグインでどのような操作をサポートするかの情報です。 CSI の仕様としては以下 4 種類のケーパビリティをサポートします。

PluginCapability
- CSI プラグイン全体でサポートする機能
- Controller Service を提供するか、オンラインボリューム拡張をサポートするかなどの情報を含む
- CO は GetPluginCapabilities メソッドで取得可能
VolumeCapability
- Volume のファイルシステムのタイプやアクセス制御（リードオンリーなのかなど）、容量などの情報を含む
- CO は CreateVolume, ControllerPublishVolume などのメソッドで指定や値の検証が可能
ControllerServiceCapability
- Create/Delete , Publish/Unpublish などの操作をサポートするかの情報を含む
- CO は ControllerGetCapabilities メソッドで取得可能
NodeServiceCapability
- Stage/Unstage などの操作をサポートするかの情報を含む
- CO は NodeGetCapabilities メソッドで取得可能

CSI の利用用途

CSI プラグインとしてはすでに、 NFS や iSCSI などの公式のサンプルプラグインを含め、 AWS や GCP のストレージサービスの利用を可能とするプラグインが存在します。以下の公式ドキュメントに既に存在する CSI プラグインに関する記述があります。

kubernetes-csi.github.io

簡単な CSI Plugin の実装を追ってみる

以上だけだと CSI プラグインの動作が把握し難いので、 kubernetes-csi の drivers リポジトリにある HostPath plugin の動作を追ってみます。このプラグインはリポジトリに kubernetes への deploy 用 yaml ファイルも同梱されており動作を追うのに便利です。

HostPath plugin では CSI の Identity, Controller, Node 三種類の gRPC サービスを提供します。その実装は github.com/kubernetes-csi/drivers/tree/master/pkg/hostpath に存在します。

Identity Server は特に特殊な実装をしていません。 github.com/kubernetes-csi/drivers/blob/master/pkg/csi-common に存在する、他の Driver と共通のデフォルトメソッド実装をそのまま使用しています。

Controller Server では CreateVolume, DeleteVolume , CreateSnapshot, DeleteSnapshot, ListSnapshots を実装します。 CreateVolume, DeleteVolume では複雑なことをしておらず、 UUID 付きの名前のディレクトリを掘って Volume に見立て、 map で Volume の管理を行います。 CreateSnapshot, DeleteSnapshot はそれに似て、 Volume のディレクトリを tar czf で固めて同じく UUID 付きのファイルに保存・管理します。

Node Server では NodePublishVolume, NodeUnpublishVolume, NodeStageVolume , NodeUnstageVolume を実装します。 NodePublishVolume, NodeUnpublishVolume は作成された Volume のディレクトリを k8s.io/kubernetes/pkg/util/mount パッケージを介して mount/unmount します。 NodeStageVolume , NodeUnstageVolume は単にリクエストパラメータ中の Volume やターゲットパスが空でないことをチェックしているだけになります。 HostPath プラグインでは他に複雑なチェックをする必要が無いためこうなっているのではと思われます。

kubernetes 上で想定される構成は以下の通りになっているようです。

Controller Server のメソッドの参照
- CreateVolume など Volume の作成・削除は https://github.com/kubernetes-csi/external-provisioner コンテナを利用
- ControllerPublishVolume など Volume の Publish 操作は https://github.com/kubernetes-csi/external-attacher を利用
- CreateSnapshot などスナップショットの操作は https://github.com/kubernetes-csi/external-snapshotter を利用
Node Server のメソッド参照
- https://github.com/kubernetes-csi/driver-registrar を使って CSIDriver を kubelet から参照可能にする
  - kubelet から CSIDriver をプラグインにという格好で読み出し参照するロジックは Kubernetes 本体リポジトリの kubelet と関連パッケージに存在

おわりに

以上、 CSI とその Kubernetes へのインテグレーションに関する記事でした。 CSI は可搬性を担保しつつ Kubernetes での利用を重要視されて作られているようですが、 CSI の仕様含めエコシステムもまだ枯れているとは言えず、機能も多くはありません。 Kubernetes 用の単なるサンプルとはいえ HostPath プラグインで tar コマンドを実行してスナップショットを作成したりなど、もう少し抽象化されてほしい箇所もまだ存在します。

しかしながら個人的には Kubernetes での柔軟なストレージ利用には、以前の記事で挙げた通りまだまだ課題があると思っており、今後の発展を期待したいと考えています。

syucream.hatenablog.jp

2018-11-05

Kubernetes 上で動作するコンテナから安全に FUSE を利用したかった

本題の通りの気持ちがあったのですが、結論としては手軽にできる良い方法は無いようでした。備忘録的に挑戦した事を記録しておきます。

背景: FUSE 利用のモチベーション

言うまでもなくファイル I/O はシステム開発においてよく使われる機構であり、多くのプログラミング言語において標準ライブラリまたはそれに類するライブラリでファイル I/O をサポートしていると思われます。

さて Kubernetes 上でマイクロサービスを実装していく事が多々あるこのご時世、各マイクロサービスで実装言語非依存で使えるデータの読み書きのロジックがあると、例えば何らかの設定ファイルの動的受け渡しやロギングにおいて便利な可能性があります。これに似た課題を抱えて解決に向かったプロダクトとして Envoy Proxy があるかと思われます。これは元々ネットワーキングやトレーシングの課題などを各言語ごとのライブラリで対応していたもののつらくなり、 HTTP や gRPC を解釈するプロキシを導入することで緩和に成功しています。

前述の通りファイル I/O は HTTP や gRPC をしゃべるより更にリーズナブルな共通プロトコルであると考えられもしそうです。さらに FUSE を使えばファイルシステムとしてインタフェースを提供して、かつ裏側で複雑なロジックを動かすこともできるように考えられます。そんなわけで、 Kubernetes 上で動作するコンテナからいい感じに FUSE で実装したファイルシステムをマウントして利用できると良いかもと着想した次第です。

FUSE 利用の限界

同じような思考をしたり、あるいは既存 FUSE 資産を利用したい人たちが多々居たようで、 Kubernetes 公式リポジトリの Issue でも議論がされていたりします。

github.com

残念ながらこの 2015 年に open した issue は 2018 年も終わりを迎える今日においても close されていません。どういう点がネックになるかというと

/dev/fuse を open するのに特権が必要
mount, umount するのに SYS_ADMIN ケーパビリティが必要

が挙げられ、一応利用できなくは無いものの利用条件を満たすため現状は privileged mode でコンテナを動かす羽目になり、やや安全面でリスクがあるように思えます。

FUSE 利用パターンいくつか

前述の Issue の中で幾つかのアイデアも提示されています。少しこれらを実際に試してみることにしました。多少このリスクを緩和するため、大きく分けて以下の方針がありそうです。

コンテナ外でなんとかするパターン
特権を持つコンテナをアプリケーションを動作させるコンテナと分離するパターン

前者は場合によっては可搬性を損なうか導入が非常に困難になる恐れがあり、また後者であればコンテナ内の世界で完結させることができそうです。従ってまずは後者のアイデアを試してみます。

以下に検証用に用意した Dockerfile や設定ファイルを配置しました。 FUSE でマウントするのは libfuse に含まれる、ファイルを read すると "Hello, World!" を返す hello.c を利用しておきます。

github.com

こちらのリポジトリでは initContainers で mount するのと、 postStart , preStop でサイドカーコンテナで mount/umount する案を試してみました。が、結局マウント先を参照するアプリケーションコンテナからはマウントした先は参照できなくなります。結局この例だと特権を渡さざるを得ませんでした。

次の一手のアイデア

今回特権を要求されたのが、コンテナ内から /dev/fuse が open できない点なので、この点を何とかできれば多少課題は緩和されるかもです。 Kubernetes は Device Plugin という機構を最近サポートしているようで、これで何とかできる可能性があるような全く無いような気がしています。また Kubernetes v1.10 からベータになったというストレージ接続インタフェース CSI は将来も見据えてこの問題に取り組むのに、もしかしたら良い機構かもしれません。

いずれにせよ Kubernetes 自体の調査をほとんどやっていないので次の一手はこれら周辺技術の調査からかと考えております。

Confluent Schema Registry とは

Confluent Schema Registry に実際に触れてみる

クイックスタート

Avro のスキーマを登録してみる

Avro のスキーマを更新してみる; 互換性がある場合

Avro のスキーマを更新してみる; 互換性がない場合

余談; _schemas topic

Confluent Schema Registry の実装を覗いてみる

avro-serializer: Schema Registry と通信した結果でシリアライズ・デシリアライズを行う

client: Schema Registry のクライアント

core: Schema Registry のコア実装

おわりに

Avro のデータ型とコード生成

基本的なデータ型表現セット GenericData

GenericData を読み書きする GenericDatumReader/GenericDatumWriter

より安全で高速な表現セット SpecificData

SpecificData なクラスのコード生成

Java 意外の言語ではどうなる？

飽きてきたので

Apache Avro はデータフォーマットとエコシステムのひとつ

Avro はスキーマの表現力が強い

Avro のスキーマを記述するのがつらい

Avro 自体の圧縮効率はそんな高くない？

Object Container Files フォーマットの便利さ

avro-tools が開発運用に便利

飽きてきたので

TL;DR

つらつら発表内容

15 分枠

GCPでStreamなデータパイプライン運用しはじめた by @shoe116

行動ログ処理基盤の構築 by @hirosassa

LT 枠

Google Cloud ML Engineに浸かってみる by @yudeayase

PoC案件が多すぎてつらいので、パイプラインを使いまわすツールを入れた。 by @mori_kaz0429

Cloud Composer & Cloud Dataflow によるバッチETLの再構築 by @yuzutas0

DigdagでETL処理をする by @nakano_shota

Comet.ml で AutoML ライブラリ開発（仮） by @Y_oHr_N

データ活用の際にハマってしまったログ・データスキーマ設計 by @yu-ya4

今後に向けて

CSI(Container Storage Interface) とは

CSI の仕様

CSI のサービス

CSI における Volume の状態遷移

CSI のケーパビリティ

CSI の利用用途

簡単な CSI Plugin の実装を追ってみる

おわりに

背景: FUSE 利用のモチベーション

FUSE 利用の限界

FUSE 利用パターンいくつか

次の一手のアイデア

余談; `_schemas` topic