PyFlink Get Started Quickly

PyFlink の概要

PyFlink は Flink のサブモジュールであり、Flink プロジェクト全体の一部です。主な目的は、Flink の Python 言語サポートを提供することです。機械学習やデータ分析などの分野では、Python 言語が非常に重要であり、最も重要な開発言語でもあります。そのため、より多くのユーザーのニーズに応え、Flink のエコシステムを拡大するために、PyFlink プロジェクトを立ち上げました。

PyFlink プロジェクトには、主に 2 つの目標があります。1 つ目は、Flink のコンピューティング能力を Python ユーザーに提供することです。つまり、Flink 内に一連の Python API を提供し、Python 言語に精通したユーザーが Flink ジョブを簡単に開発できるようにします。

2 つ目は、Python エコシステムを Flink 上で分散実行できるようにすることです。Flink 内に Python 用の API を提供しますが、ユーザーにとって Python API の使い方や各 API の用途を学ぶ必要があるため、学習コストがかかります。そこで、ユーザーが普段使い慣れている Python ライブラリの API をそのまま利用できるようにし、裏側の計算エンジンとして Flink を使用することで、学習コストを削減したいと考えています。これは今後取り組んでいく予定であり、現在は初期段階にあります。

以下の図は、PyFlink プロジェクトの開発状況を示しています。現在までに 3 つのバージョンがリリースされており、サポート内容はますます充実してきています。

PyFlink の関連機能の紹介

ここでは、PyFlink の主な機能として、Python Table API、Python UDF、ベクトル化 Python UDF、Python UDF メトリクス、PyFlink 依存関係管理、および Python UDF 実行最適化について紹介します。

Python Table API

Python Table API の目的は、ユーザーが Python 言語で Flink ジョブを開発できるようにすることです。Flink には、Process API、Function API、Table API という 3 種類の API があります。前者 2 つは比較的低位な API で、Process API と Function API をベースに開発されたジョブは、ユーザーが定義したロジックをそのまま厳密に実行します。一方、Table API は比較的高位な API で、Table API をベースに開発されたジョブは、一連の最適化を経てから実行されます。

Python Table API は、その名の通り、Table API に Python 言語のサポートを提供するものです。

以下は、Python Table API で開発された Flink ジョブの例です。このジョブのロジックは、ファイルを読み取り、ワードカウントを計算し、計算結果をファイルに書き込むというものです。シンプルな例ですが、Python Table API ジョブ開発の基本的なフローがすべて含まれています。

まず、ジョブの実行モード（バッチモードかストリームモードか）、同時実行数、ジョブの設定などを定義する必要があります。次に、ソーステーブルと結果テーブルを定義します。ソーステーブルはジョブのデータソースの場所とデータ形式を定義し、結果テーブルはジョブの実行結果の出力先とデータ形式を定義します。最後に、ジョブの実行ロジックを定義します。この例では、ワードカウントを計算します。

以下は Python Table API のいくつかのスクリーンショットです。API の数と機能が比較的充実していることがわかります。

Python UDF

Python Table API はリレーショナルな API であり、その機能は SQL に匹敵します。SQL におけるカスタム関数は、SQL の利用範囲を大幅に拡張できる非常に重要な機能です。Python UDF の主な目的は、ユーザーが Python 言語でカスタム関数を開発できるようにすることで、Python Table API の利用シーンを拡大することです。同時に、Python UDF は Python Table API ジョブだけでなく、Java Table API ジョブや SQL ジョブでも使用できます。

PyFlink では、複数の方法で Python UDF を定義できます。ScalarFunction を継承した Python クラスを定義することも、通常の Python 関数やラムダ関数を定義してカスタム関数のロジックを実装することもできます。また、Callable Function や Partial Function を通じた Python UDF の定義もサポートしています。ユーザーは必要に応じて最適な方法を選択できます。

PyFlink では、Python Table API、Java Table API、SQL など、Python UDF を使用する複数の方法を提供しています。順を追って説明します。

Python Table API で Python UDF を使用する場合、Python UDF を定義した後、まず Python UDF を登録する必要があります。テーブル環境の register を呼び出して登録し、名前を付けると、その名前を使ってジョブ内で Python UDF を使用できるようになります。

Java Table API でも使い方は似ていますが、登録方法が異なります。Java Table API ジョブでは、DDL ステートメントを使用して登録する必要があります。

また、SQL ジョブでも Python UDF を使用できます。前述の 2 つの方法と同様に、まず Python UDF を登録する必要があります。SQL スクリプト内の DDL ステートメント、または SQL Client の環境設定ファイルを通じて登録できます。

Python UDF アーキテクチャ

Python UDF の実行アーキテクチャについて簡単に説明します。Flink は Java 言語で書かれており、Java 仮想マシン上で動作します。一方、Python UDF は Python 仮想マシン上で動作するため、Java プロセスと Python プロセスが相互に通信する必要があります。さらに、状態、ログ、メトリクスも両プロセス間で伝送する必要があり、その伝送プロトコルは 4 種類をサポートする必要があります。

ベクトル化 Python UDF

ベクトル化 Python UDF の主な目的は、Pandas や NumPy など、データ分析分野でよく使用される Python ライブラリを活用して、Python ユーザーが高性能な Python UDF を開発できるようにすることです。

ベクトル化 Python UDF は通常の Python UDF と対比されるもので、両者の違いは以下の図で確認できます。

以下の図は、ベクトル化 Python UDF の実行フローを示しています。まず Java 側で、複数行のデータを Arrow 形式に変換して保存し、Python プロセスに送信します。Python プロセスはデータを受信した後、Pandas のデータ構造に変換してから、ユーザー定義のベクトル化 Python UDF を呼び出します。同時に、ベクトル化 Python UDF の実行結果は Arrow 形式のデータに変換され、Java プロセスに送り返されます。

使い方は通常の Python UDF と似ていますが、いくつかの違いがあります。まず、ベクトル化 Python UDF の宣言方法では、UDF タイプを追加してベクトル化 Python UDF であることを宣言する必要があり、UDF の入出力タイプは Pandas Series になります。

Python UDF メトリクス

前述の通り、Python UDF を定義する方法は複数ありますが、Python UDF でメトリクスを使用する場合は、ScalarFunction を継承して定義する必要があります。Python UDF の open メソッドには Function Context パラメーターが提供されます。ユーザーは Function Context パラメーターを通じてメトリクスを登録し、登録したメトリクスオブジェクトを通じてレポートできます。

PyFlink 依存関係管理

PyFlink の依存関係には、主に通常の PyFlink ファイル、アーカイブファイル、サードパーティライブラリ、PyFlink インタープリター、Java Jar パッケージなどの種類があります。各タイプの依存関係に対して、PyFlink は API ベースとコマンドラインオプションベースの 2 つのソリューションを提供しており、いずれかを選択できます。

Python UDF 実行最適化

Python UDF の実行最適化には、実行計画の最適化とランタイム最適化の 2 つの側面があります。SQL とよく似ており、Python UDF を含むジョブはまず事前定義されたルールに基づいて最適な実行計画を生成します。実行計画が確定した後、実際の実行時にさらなる最適化手法を適用することで、最大限の実行効率を実現します。

Python UDF 実行計画の最適化

実行計画の最適化には、主に以下の 3 つのアプローチがあります。1 つ目は、異なるタイプの UDF の分割です。1 つのノードに複数タイプの UDF が同時に含まれる場合があるため、異なるタイプの UDF を一緒に実行できません。2 つ目はフィルタープッシュダウンで、Python UDF ノードへの入力データ量を可能な限り削減し、ジョブ全体の実行パフォーマンスを向上させることが主な目的です。3 つ目は Python UDF チェイニングで、Java プロセスと Python プロセス間の通信オーバーヘッドおよびシリアル化とデシリアル化のオーバーヘッドは比較的大きいため、Python UDF チェイニングによって Java プロセスと Python プロセス間の通信オーバーヘッドを最小化できます。

異なるタイプの UDF の分割

あるジョブに add（Python UDF）と subtract（ベクトル化 Python UDF）という 2 つの UDF が含まれているとします。デフォルトでは、このジョブの実行計画には 1 つの Project ノードがあり、両方の UDF が同時にこの Project ノードに配置されます。この実行計画の主な問題は、通常の Python UDF がデータを 1 件ずつ処理するのに対し、ベクトル化 Python UDF は複数件をまとめて処理するため、このままでは実行できないことです。

しかし、分割により、この Project ノードを 2 つの Project ノードに分割できます。最初の Project ノードには通常の Python UDF のみを含め、2 番目のノードにはベクトル化 Python UDF のみを含めます。異なるタイプの Python UDF を異なるノードに分割することで、各ノードには 1 種類の UDF のみが含まれるため、オペレーターは UDF のタイプに応じて最適な実行方法を選択できます。

Python UDF の前へのフィルタープッシュダウン

フィルタープッシュダウンの主な目的は、フィルターオペレーターを Python UDF ノードの前にプッシュダウンし、Python UDF ノードへのデータ量を最小化することです。

あるジョブの元の実行計画に、add と subtract の 2 つの Project ノードと、1 つの Filter ノードが含まれているとします。この実行計画は機能しますが、最適化が必要です。Python ノードが Filter ノードの前に配置されているため、Filter の前に Python UDF の計算が行われていますが、Filter を Python UDF の前にプッシュダウンすれば、Python UDF ノードへの入力データ量を大幅に削減できます。

Python UDF チェイニング

あるジョブに add と subtract という 2 つの通常の Python UDF が含まれているとします。実行計画には 2 つの Project ノードがあり、最初の Project ノードで subtract を計算してから、2 番目の Project ノードに転送して実行されます。

主な問題は、subtract と add が 2 つの異なるノードに配置されているため、計算結果を Python から Java に送り返し、さらに Java プロセスから 2 番目のノードの Python に送信する必要があることです。つまり、データが Java プロセスと Python プロセスの間を往復するため、完全に不要な通信オーバーヘッドとシリアル化とデシリアル化のオーバーヘッドが発生します。そこで、実行計画を最適化し、add ノードと subtract ノードを 1 つのノード内で実行し、subtract ノードの結果が出たら直接 add ノードを呼び出すようにします。

Python UDF ランタイム最適化

現在、Python UDF 操作の実行効率を向上させる方法は主に 3 つあります。1 つ目は Cython 最適化で、Python コードの実行効率を向上させるために使用されます。2 つ目はシリアル化とデシリアル化の効率改善です。3 つ目はベクトル化 Python UDF 機能の提供です。

PyFlink の関連機能のデモ

まず、PyFlink のデモが用意されているこのページを開きます。これらのデモは Docker で実行されるため、デモを実行するにはローカルに Docker 環境をインストールする必要があります。

次に、コマンドを実行します。これにより PyFlink クラスターが起動し、後で実行する PyFlink のサンプルはこのクラスターに送信されて実行されます。

最初のサンプルはワードカウントです。まず、環境、ソース、シンクなどを定義してから、ジョブを実行できます。

これがジョブの実行結果です。Flink という単語が 2 回、PyFlink という単語が 1 回出現していることがわかります。

次に、Python UDF のサンプルを実行します。このサンプルは前のものとやや似ています。まず、PyFlink を使用し、バッチモードで実行し、ジョブの同時実行数は 1 と定義します。異なる点は、ジョブ内で UDF を定義し、その入力には Bigint 型の 2 列を含み、出力タイプもそれに対応しています。この UDF のロジックは、2 つの列の合計を結果として出力します。

ジョブを実行すると、実行結果は 3 です。

次に、依存関係を含む Python UDF を実行します。前のジョブの UDF には依存関係が含まれておらず、2 つの入力列を直接合計していました。このサンプルでは、UDF がサードパーティの依存関係を参照しており、set python requirement API を通じて設定できます。

次にジョブを実行すると、実行結果は前と同じです。2 つのジョブのロジックが同じだからです。

次に、ベクトル化 Python UDF のサンプルを見ていきます。UDF を定義する際に UDF タイプのフィールドを追加し、ベクトル化 Python UDF であることを示します。その他のロジックは通常の Python UDF と同様です。最終的な実行結果も 3 で、ロジックは前と同じく 2 つの列の合計を計算するためです。

次に、Java の Table ジョブで Python を使用するサンプルを見ていきます。このジョブでは、DDL ステートメントで登録した Python UDF を使用し、execute SQL ステートメント内で使用します。

次に、純粋な SQL ジョブで Python UDF を使用するサンプルを見ていきます。リソースファイルで、add1 という名前の UDF を宣言し、そのタイプは Python で、UDF の場所も確認できます。

次に実行すると、実行結果は 234 です。

PyFlink の今後の計画

現在、PyFlink は Python Table API のみをサポートしています。次期バージョンでは DataStream API のサポートを追加する予定であり、Python UDAF と Pandas UDAF もサポートする予定です。さらに、実行層での PyFlink の実行効率の最適化も継続して取り組んでいきます。