Apache Flink トークシリーズ - PyFlink のコアテクノロジーを公開

皆さんこんにちは。本日のサミットで Apache PyFlink のコア技術を皆さんと共有できることを大変嬉しく思います。

まず最初に、簡単に自己紹介をさせてください。私の名前は Sun Jincheng、愛称は Jinzhu で、Alibaba 出身です。2016 年からオープンソースの構築に携わっています。現在は Apache Flink の PMC メンバー、Apache Beam および Apache IoTDB の PMC メンバーです。彼は Apache Software Foundation のメンバーでもあります。また、技術的なブログを書いたり、ビデオ コースを録画したりすることも好きです。私の公式アカウントをフォローしていただければ幸いです。

今日は4つのパートに分けてお伝えします。まずはPyFlinkのミッションとビジョンを簡単に見ていき、次にPyFlinkのコアテクノロジーの紹介に焦点を当て、最後にPyFlinkの今後の計画と既存のアプリケーション事例を簡単に紹介します。それでは、今日の最初の部分である、PyFlink の使命とビジョンから始めましょう。

まず、Apache Flink はステートフルな分散ストリーミング コンピューティング フレームワークです。有限および無限のデータ セットを操作できます。

有限および無限のデータ セットに対するストリーム コンピューティング処理には、業界に 2 つの一般的なアーキテクチャがあります。1 つは、ストリームをバッチの特殊なケースとして扱うマイクロバッチ モードです。もう 1 つは、バッチをストリームの特殊なケースと見なす純粋なストリーム アーキテクチャ モードである Apache Flink アーキテクチャ モードです。純粋なフロー設計により、計算の遅延が最小限に抑えられます。

では、Flink の分散型 1 秒未満のレイテンシ機能はどのようにユーザーに公開されるのでしょうか? Flink は、ユーザーが選択できる SQL、DataStream、および ProcessFunction のマルチレイヤー API を提供しますが、残念ながら Java ユーザーのみに提供されます。

したがって、Flink の機能を拡張し、より多くのユーザー グループにサービスを提供することは非常に有意義です。では、Flink で複数の言語をサポートするにはどうすればよいでしょうか?どの言語がサポートされますか？

PyFlink に取り組む前に調査を行ったところ、2020 年には Python が Java よりも活発になり、この傾向は引き続き上昇していることがわかりました。

それでは、Python がなぜ人気があるのか​​、そして Python を使ってどのような仕事をしているのかを詳しく見ていきましょう。これらの問題を抱えながら、AI の台頭により、Python はデータ分析や Web 開発だけでなく、AI/機械学習でも広く使用されるようになりました。さらに興味深いのは、高速道路パトロールの警官でさえ趣味を Python プログラミングに変えており、8 歳や 9 歳の子供たちも Python を使用して楽しいゲームを作っていることです。これだけでも Python の人気がわかります。そのため、Flink の多言語サポートでは、最も重要な開発言語として Python が使用されます。

Python は非常に人気があり、非常に成熟したエコシステムを持っています。しかし、ここには典型的な問題があります。これらのエコシステム ライブラリのほとんどはスタンドアロン モードです。今日のビッグ データ時代において、Python エコシステムが直面している典型的な問題は次のとおりです。

膨大なデータの処理をどのようにサポートし、分散機能をどのように提供するのでしょうか?

したがって、Flink の機能はより多くのユーザー グループに向ける必要があり、Python が最も人気のある言語であるため、Python は複数の言語で Flink によってサポートされる最初の言語になりました。同時に、Python 言語の分散機能の欠如に直面し、PyFlink の使命は、Python エコシステムに分散処理機能を備えることです。したがって、Pyflink の使命は、Flink 機能を Python ユーザーにエクスポートし、Python エコシステムが分散機能を持てるようにすることです。

さて、PyFlink がどのようにその使命を達成し、どのようなコアな技術的詳細を持っているかを見てみましょう。

まず第一に、Flink 機能を Python ユーザーにエクスポートする際の中心的な問題は、明らかに Python VM と Java VM 間のハンドシェイクです。それらの間の通信を確立する必要があり、これが PyFlink が解決する主な問題です。

PVM と JVM 間の通信問題を解決するために、Py4J を選択し、PythonVM でゲートウェイを起動し、Java VM でゲートウェイ サーバーを起動して Python リクエストを受け入れました。同時に、TableENV、Table など、Java API と同じオブジェクトを Python API に提供しました。このように、Python が Python API を記述する場合、本質的には Java API を呼び出していることになります。同時に、ジョブのデプロイメントの問題もあります。Python コマンド、Python シェル、CLI など、複数の方法でジョブを送信できます。

では、Py4J と JVM 間の相互作用の原理は何でしょうか?実際、コアとなる仕組みは、Python 側でオブジェクトが作成されるたびに、Java 側で対応する Java オブジェクトが作成され、オブジェクト ID が生成されるというものです。Java 側では、オブジェクト ID とオブジェクトを保存するために Map を使用します。同時に、オブジェクト ID が Python 側に返されます。オブジェクト ID とメソッド パラメータに基づいて Python 側で実行される操作は、基本的に Java オブジェクトに対する操作です。

では、このようなアーキテクチャの利点は何でしょうか? 1つ目は、シンプルさと、Python APIセマンティクスとJava APIの一貫性の確保です。2つ目は、PythonジョブがJavaと同じ極限のパフォーマンスを実現できることです。先日終了したアリババダブル11カーニバルでは、ピーク処理能力40億が生み出されました。

さて、既存の Flink 関数を Python ユーザーにエクスポートしたので、Python エコシステムの関数を Flink に導入し、Python 関数を配布する方法を引き続き検討してみましょう。どうやってそれを達成するのでしょうか? Flink Table API の現在の状態と既存の Python クラス ライブラリの特性を組み合わせると、既存のすべての Python クラス ライブラリ関数をユーザー定義関数 (UDF) として考え、Flink に統合することができます。このように、Python エコシステムを Flink に統合する方法は、それを UDF として扱うことであることがわかりました。では、統合の核心的な問題は何でしょうか?はい、それは Python UDF の実行の問題です。さて、この核心的な問題にどう対処すればいいのでしょうか?

Python UDF 実行問題を解決するには、VM 間の通信だけの問題ではありません。Python 実行環境の管理、Java と Python 間の業務データの解析、Flink State Backend 機能の Python への出力、Python UDF 実行の監視などが関係しており、非常に複雑な問題です。このような複雑な問題に直面して、私たちは統一プログラミングモデル Apache Beam を選択しました。多言語とマルチエンジンのサポートの問題を解決するために、Beam は Portability Framework と呼ばれるアーキテクチャを高度に抽象化しました。次の図に示すように、Beam は現在 Java/Go/Python などの複数の言語をサポートしています。エンジンと UDF 実行環境の問題は、図の下部にある Beam Fu Runners と Execution の間で解決されます。その中核は、Protobuf を使用してデータ構造を抽象化し、通信に gRPC プロトコルを使用し、コア gRPC サービスをカプセル化することです。つまり、現時点では、Beam はホタルのような存在であり、PyFlink が UDF 実行の問題を解決する道を照らしています。次に、Beam が提供する gRPC サービスについて見てみましょう。

図に示すように、Runner 部分が Java オペレータ実行、SDK Worker 部分が Python 実行環境です。Beam では、Control/Data/State/Logging などのサービスが抽象化されています。そして、これらのサービスは Beam の Flink ランナー上で長期間にわたって安定して効率的に実行されてきました。つまり、PyFlink UDF 実行の巨人たちの肩の上に立つことができるということです :)。ここで、Apache Beam には API レベルと UDF 実行レベルの両方でソリューションがあることがわかります。PyFlink は、API レベルでの VM 通信の問題を解決するために Py4J を使用し、UDF 実行要件に対する UDF 実行環境の問題を解決するために Beam の Portability Framework を使用します。これは、PyFlink が技術選択において最低コストで設定された目標を達成するという原則を厳格に遵守し、PyFlink の長期的な開発に最適な最も適切な技術アーキテクチャを常に選択することを示しています。

さて、Flink はどのようにして複数の言語をサポートするのでしょうか?

API レベルでは、algin の既存の Java 言語 API を他の言語で処理する必要があります。

言語実行環境に関しては、Flink は Beam が提供するインフラストラクチャを再利用できます。つまり、Flink ランナーと fnapi レベルで基本的なサービスとデータ構造を簡単に再利用できます。これにより、Flink は複数の言語を簡単にサポートできるようになります。

PyFlink の UDF アーキテクチャ設計を見てみましょう。

それでは、PyFlink UDF の全体的なアーキテクチャを見てみましょう。 UDF アーキテクチャでは、Java VM と Python VM 間の通信、およびコンパイル フェーズとランタイム フェーズでのさまざまな要件を考慮する必要があります。

図では、緑色で Java VM の動作を表し、青色で Python VM の動作を表しています。

まず、コンパイル段階、つまりローカル設計を見てみましょう。ローカル設計は純粋な API マッピング呼び出しであり、通信の問題を解決するには Py4J が必要です。つまり、図に示すように、Python が API を実行するたびに、Java 内の対応する API が同期的に呼び出されます。 UDF をサポートするには、UDF 登録 API である register_function を追加する必要があります。ただし、登録だけでは不十分です。ユーザーが Python UDF をカスタマイズする場合、サードパーティのライブラリに依存することがよくあります。

したがって、依存関係を追加するメソッド、つまり add_Python_file() などの一連の追加メソッドも追加する必要があります。

Python ジョブを記述するときに、Java API も同時に呼び出されます。ジョブを送信する前に、Java 側で .JobGraph を構築します。次に、CLI などのさまざまな方法でジョブをクラスターに送信して実行します。

実行時の Python と Java の異なる分業を見てみましょう。まず、Java 側では、通常の Java ジョブと同様に、JobMaster がジョブを TaskManager に割り当て、TaskManager がタスクを 1 つずつ実行します。タスクには、Java および Python 演算子の実行が含まれます。

Python UDF オペレーターでは、ビジネス データ通信を完了するための DataService、Python UDF の Java Statebackend への呼び出しを完了するための StateService、そしてもちろん Logging や Metrics などの他のサービスなど、Java VM と Python VM 間のさまざまな通信を完了するためのさまざまな gRPC サービスを設計します。これらのサービスはすべて Beam の Fn API に基づいて構築されており、最終的には Python ワーカーでユーザーの UDF を実行します。実行が完了すると、対応する gRPC サービスを使用して、Java 側の Python UDF オペレーターに結果が返されます。

もちろん、Python のワーカーはプロセス モードだけではなく、Docker モードや外部サービス クラスターでも実行できます。この拡張メカニズムは、PyFlink と Python エコシステム内の他のフレームワークとのその後の統合のための強固な基盤を築きます。

ここで最も重要なことは、ビーム インフラストラクチャを使用して Python UDF を実行する方法です。 pyflink が Beam の移植性フレームワークを統合して Python UDF を実行する方法を見てみましょう。入力データに対して一連の変換が実行され、その結果が別の外部ストレージ システムに書き込まれるシナリオ。 Flink は Java で開発されていますが、ユーザー定義の変換ロジックは Python で開発されています。例に示すように、ParDo が Python UDF を使用すると仮定すると、Beam に ExecutableStage が導入されます。これには、入力/出力データ型、ユーザー定義関数のペイロード、ユーザー定義関数で使用される状態とタイマーなど、ユーザー定義 Python 関数の必要な情報がすべて含まれます。同時に、Beam は特定の言語の実行環境を管理するために使用できる Java ライブラリも提供します。 「forStage()」は、SDK ハーネス部分である ExecutableStage に定義された情報に基づいて、ユーザー定義関数を実行するために必要なプロセスを生成し、ランナーと SDK ハーネス間の通信接続を確立します。

Beam の SDK ハーネスは、ParDo、Flatten などの複数の関数の実行をサポートしています。関数ごとに実行モードが異なるため、SDK ハーネスは関数を実行するための特定の操作クラスを定義します。しかし、Beam 内の各関数の実行ロジックを明確に定義するにはどうすればよいでしょうか? Beam は非常に柔軟なプラグイン メカニズムを提供します。つまり、Input/output/parDo などの関数の種類ごとに URN を定義します。このようなプラグイン メカニズムは、Flink が Beam フレームワークを統合するための便利な方法も提供します。

PyFlink で Python SDK Harness を使用する動作原理は次のとおりです。

起動フェーズ中に、Python SDK ハーネスはすべての組み込み操作の URN と操作マッピングを確立します。

新しいパッケージを処理する初期化フェーズ中に、ランナーは URN を関数とともに SDK ハーネスに送信します。 SDK ハーネスは、指定された URN に基づいて対応する操作を構築できます。この操作は、入力データと対応するユーザー定義関数ロジックを実行するために使用されます。

図に示すように、入力/出力、コーダなど、さまざまな URN を定義します。

さて、URN の登録も非常に簡単です。カスタム操作とコーダーを作成するための関数をいくつか追加するだけです。これらの関数は、Beam の Python SDK ツールキットで定義されたデコレータで装飾されます。デコレータには、URN と protobuf に基づくカスタム パラメータの 2 つのパラメータが含まれています。

さて、Python UDFをサポートした後、PandasもPyFlinkに統合しました。Pandas UDFをPyFlinkで簡単に定義できます。同時に、FlinkとPandas間の操作変換をサポートするために、frompandasとtopandas APIも提供しています。

同時に、UDF の実行パフォーマンスも継続的に最適化しています。バージョン 1.11 では、バージョン 1.10 と比較して 30 倍のパフォーマンス向上が実現されています。

さて、PyFlink の今後の計画について簡単に見てみましょう。

PyFlink の開発は、常に当初の意図に基づいて進められるべきです。私たちの目標は、既存の Flink 関数を Python ユーザーにエクスポートし、Python エコシステムの関数を Flink に統合することです。

まず、Python VM と Java VM 間の通信の問題を解決します。

次に、既存の Table API 機能を Python ユーザーに公開し、Python Table API を提供します。

これは Flink 1.9 で行われることです。

Python 関数を Flink に統合するための準備として必要な次のステップは、Apache Beam を統合して Python UDF の実行環境を提供することです。

そして、他のライブラリの依存関係に対するPythonの管理機能を強化し、

ユーザー定義関数のインターフェース定義を提供し、Python UDF をサポートします。

これが Flink 1.10 の機能です。

Python エコシステムの分散機能をさらに拡張するために、PyFlink は Pandas Series と DataFrame のサポートを提供します。つまり、ユーザーは PyFlink で Pandas UDF を直接使用できるようになります。

同時に、ユーザーの使いやすさを向上させ、ユーザーが PyFlink をより多くの方法で使用できるようにするために、将来的には Sql クライアントでの Python UDF の使用が追加される予定です。

Python ユーザーの機械学習の問題に対処するには、Python の ML パイプライン API を追加します。

Python UDF の実行を監視することは実際の運用業務にとって重要なので、PyFlink では Python UDF のメトリック管理を追加します。これは Flink 1.11 での作業です。

同時に、パフォーマンスを継続的に最適化し、PyFlink 1.12 で提供される Datastream と k8s のサポートも提供する必要があります。

今後もFlinkの既存機能をPythonエコシステムにプッシュし続け、Pythonエコシステムの強力な機能をFlinkに継続的に統合することで、Pythonエコシステムを配布するという当初の意図を完成させていきます。

もちろん、PyFlink は Zeppelin、Jupyter、PyAlink などとの統合を推進するなど、エコロジカルな統合にも注力していきます。

最後に、PyFlink の使用例をいくつか簡単に見てみましょう。

PyFlinkは、イベント駆動型/データ分析/ETL/機械学習など、さまざまなシナリオに適用できます。現在も本番環境でご利用いただいているユーザー様も多数いらっしゃいます。

例えば、Bitcoin Mainland、Jumeiなど。現在、PyFlink は成熟しており、分散コンピューティング システムを迅速に構築するためのエントリー開発言語として Flink を選択するのに非常に適しています。

現時点では、PyFlink の機能はどんどん充実しつつあり、もちろんまだまだやるべきことはありますが、いずれにしても、今後徐々に成熟していくと信じています! 2020年12月からIoT分野に力を入れ、新たな探求を始めます〜

著者について

Sun Jincheng、51CTO コミュニティ エディター、Apache Flink PMC メンバー、Apache Beam コミッター、Apache IoTDB PMC メンバー、ALC Beijing メンバー、Apache ShenYu メンター、Apache Software Foundation メンバー。ストリーム コンピューティングと時系列データ ストレージの技術分野に重点を置いています。