Flink Ecosystems

Flink SQL が外部システムと連携する仕組み

Flink SQL の原理を説明する前に、まずなぜ Flink SQL を使うのかについて考えてみましょう。
SQL は標準化されたデータクエリ言語であり、Flink SQL では Catalog を介してさまざまなシステムと連携できます。
また、豊富な組み込みオペレーターや関数が用意されており、バッチデータとストリームデータの両方を同時に処理できるため、データ分析の効率を大幅に向上させることができます。


では、なぜ Flink SQL は外部システムと連携する必要があるのでしょうか。
Flink SQL 自体はストリーム計算エンジンであり、データを保持しません。
したがって、すべてのデータは外部システムに保存されており、すべてのテーブルも外部システム上に存在します。
実際にデータの読み取りと書き込みを行うには、これらの外部システムと接続する必要があります。


Flink SQL が外部システムとどのように連携するかを説明する前に、Flink 内部での DataStream と Table の変換について見ていきましょう。
既存の DataStream プログラムがあると仮定すると、それを Table に変換して、Flink SQL の強力なクエリ機能を活用できます。
この仕組みは次の例で理解でき、Flink SQL の内部接続の仕組みと似ています。


コネクタ

Flink SQL では、外部システムと接続するコンポーネントを Connector と呼びます。
次の表は、Flink SQL がサポートする代表的なコネクタを示しています。
たとえば、Filesystem はファイルシステムに接続し、JDBC は外部のリレーショナルデータベースに接続します。
各 Connector は主に source と sink の実装を担当します。
source は外部システムからのデータ読み取りを担当し、sink は外部システムへのデータ書き込みを担当します。


フォーマット

Format は外部システム内のデータ形式を指定します。
たとえば Kafka テーブルの場合、データが CSV 形式または JSON 形式で保存されている可能性があります。
そのため、外部テーブルに接続する Connector を指定する際、通常は Format も併せて指定し、Flink がデータを正しく読み書きできるようにする必要があります。


カタログ

Catalog は外部システムのメタデータに接続し、そのメタデータ情報を Flink に提供することで、Flink が外部システム上の既存テーブルやデータベースに直接アクセスできるようにします。
たとえば、Hive のメタデータは Hive Metastore に保存されているため、Flink が Hive テーブルにアクセスするには HiveCatalog を使用してメタデータに接続します。
さらに Catalog は、Flink 自身のメタデータの永続化にも役立ちます。
たとえば HiveCatalog は、Flink が Hive にアクセスできるだけでなく、Flink が作成したテーブル情報を保存できるため、セッションを開始するたびにテーブルを再作成する必要がなく、Hive Metastore に保存されたテーブルを直接読み取れば済みます。


外部コネクタを指定してテーブルを作成するにはどうすればよいでしょうか。
次の例は、DDL を使ってテーブルを作成するものです。
これは比較的標準的な CREATE TABLE 文で、コネクタに関連するすべてのパラメータは WITH 句で指定します。
たとえば、コネクタの種類として Kafka を指定します。


DDL でテーブルを作成した後、Flink でこのテーブルはどのように使われるのでしょうか。
ここで重要な概念が Table Factory です。
DDL でテーブルを作成するか、Catalog を介して外部システムからテーブルを取得し、Catalog Table オブジェクトに変換します。
SQL 文で Catalog Table を参照すると、Flink はそのテーブルに対応する source または sink を作成します。
この source と sink を作成するモジュールが Table Factory と呼ばれます。


Table Factory の取得方法は 2 つあります。
1 つは Catalog があらかじめ Table Factory にバインドされている場合、もう 1 つは Java SPI を介して Table Factory を決定する方法です。
ただし SPI で検索する際は、候補がちょうど 1 つだけ見つからないとエラーになります。


Flink SQL でよく使われるコネクタ

Kafka コネクタ

Kafka Connector は最もよく使用されます。
Flink はストリーム計算エンジンであり、Kafka は最も人気のあるメッセージキューであるため、Flink を使用するユーザーの多くは Kafka も併用しています。
Kafka テーブルを作成するには、connector フィールドを Kafka に指定し、Kafka に対応する Topic などを指定する必要があります。
これらのパラメータとその意味は次の図で確認できます。


Kafka Connector を使用するには、Kafka の依存 Jar パッケージを追加する必要があります。
使用する Kafka のバージョンによって、追加する Jar パッケージも異なり、これらのパッケージは公式サイトからダウンロードできます。


Elasticsearch コネクタ

Elasticsearch Connector は Sink のみを実装しているため、ES への書き込みはできますが、読み取りはできません。
コネクタタイプは ES6 または ES7 を指定できます。
Hosts は各 ES ノードをドメイン名とポート番号の形式で指定します。
Index は ES の書き込み先インデックスを指定し、従来のデータベースのテーブルに相当します。
Document Type は従来のデータベースのテーブル内の行に似ていますが、ES7 では指定不要です。


ES の Sink は append モードと upsert モードの 2 つをサポートしています。
ES テーブルの定義時に主キーを指定した場合、Sink は upsert モードで動作します。
主キーを指定しない場合は append モードで動作しますが、ROW 型や MAP 型などは主キーとして使用できません。


同様に、ES を使用する場合も追加の依存関係を指定し、ES のバージョンに応じた ES Connector を追加する必要があります。


ファイルシステムコネクタ

この Connector はファイルシステムに接続し、そのファイルシステム上のファイルを読み書きします。
ここでいう FileSystem は Flink の FileSystem 抽象化を指し、ローカルファイルシステム、Hadoop、S3、OSS など、さまざまな実装をサポートしています。
同時にパーティションもサポートしており、Hive と同様のパーティションディレクトリ構造を採用していますが、パーティション情報を Catalog に登録する必要はありません。


Hive コネクタ

Hive は最も初期の SQL エンジンの 1 つで、ほとんどのユーザーはバッチ処理シナリオで使用しています。
Hive Connector は 2 つのレベルに分けられます。
まずメタデータ接続では、HiveCatalog を使用して Hive メタデータに接続します。
同時に、HiveTableSource と HiveTableSink を提供して、Hive テーブルデータの読み取りと書き込みを行います。


Hive Connector を使用するには、Hive Catalog を指定する必要があります。
次の例は Hive Catalog の指定方法を示しています。


Hive Connector の使用にも追加の依存関係が必要です。
使用する Hive のバージョンに応じて、対応する Jar パッケージを選択できます。


外部システムと接続するコネクタのほかに、組み込みコネクタも用意されています。
これらは一方では新規ユーザーが Flink SQL の強力な機能をいち早く体験できるように支援し、他方では Flink 開発者のコードデバッグにも役立ちます。


DataGen コネクタ

DataGen Connector はデータ生成機能です。
たとえば、ここで DataGen テーブルを作成し、いくつかのフィールドを指定します。
コネクタタイプを DataGen に指定してこのテーブルを読み取ると、Connector がデータの生成を担当します。
つまり、データはどこかにあらかじめ保存されているのではなく、その場で生成されます。
ユーザーは DataGen Connector に対して詳細な制御を行うことができ、たとえば 1 秒あたりに生成するデータ行数を指定したり、フィールドをシーケンス生成（小さい値から大きい値へ）またはランダム生成に指定したりできます。


Print コネクタ

Print Connector は Sink 機能を提供し、すべてのデータを標準出力または標準エラー出力に出力します。出力形式は、先頭にロウカインドが付加されます。Print テーブルを作成する際は、コネクタタイプを print に指定するだけです。

BlackHole コネクタ

BlackHole Connector も Sink の 1 つで、すべてのデータを破棄します。つまり、データを書き込んでも何も処理されず、そのまま破棄されます。主にパフォーマンステストに使用されます。BlackHole を作成するには、コネクタタイプを BlackHole に指定するだけです。