How to migrate Hive SQL to Flink SQL

関連タグ：1.Realtime Compute for Apache Flink
2. Flink Python

1、Hive SQL 移行の動機

Flink はストリームコンピューティングのデファクトスタンダードとして確立されており、現在、国内外のリアルタイムコンピューティングやストリームコンピューティングでは Flink と Flink SQL が主流の選択肢となっています。さらに、Flink はストリーミングバッチビッグデータ計算エンジンとしても広く知られています。

ただし、Flink にも課題があります。たとえば、現在の大規模アプリケーションは主にストリームコンピューティングに集中しており、Flink バッチコンピューティングの活用はまだ限定的です。真の意味でのストリームとバッチの統合をさらに推進するには、業界でより多くの Flink バッチコンピューティングを推進し、既存のオフラインエコシステムをより積極的に取り込む必要があります。現在、業界のオフラインエコシステムは Hive が中心です。そのため、過去のバージョンでは Hive Catalog、Hive 構文互換性、Hive UDF 互換性、ストリーミング Hive など、Hive 関連の統合機能を数多く追加してきました。Flink 1.16 では、HiveSQL の互換性をさらに改善し、HiveServer2 のプロトコル互換性もサポートしました。

では、なぜ Flink は Hive SQL の移行をサポートするのでしょうか。一方では、Hive を中心としたオフラインデータウェアハウスユーザーを惹きつけ、バッチ計算エンジンを継続的に改善し、主流のバッチ計算エンジンとして位置づけるためです。他方では、Hive SQL との互換性により、既存のオフラインユーザーが Flink を使ってオフラインビジネスを開発する際のハードルを下げることができます。さらに、エコシステムはオープンソース製品にとって最大のハードルです。Flink は既にリアルタイム分野では非常に豊富なエコシステムツールを備えていますが、オフライン分野のエコシステムはまだ不足しています。Hive エコシステムとの互換性により、Hive のオフラインエコシステムツールやプラットフォームを迅速に統合し、ユーザーの導入コストを削減できます。最後に、これはストリームとバッチの統合を実現する重要な要素でもあります。業界にストリームとバッチの統合コンピューティングエンジンを導入し、さらに統合されたストリームバッチ SQL の実現を推進したいと考えています。

ユーザーの視点から見ると、なぜ Hive SQL を Flink SQL に移行すべきでしょうか。

プラットフォーム側では、統合ストリームバッチ計算エンジンにより Flink エンジン一套の保守だけでよくなり、メンテナンスコストを削減し、チームの開発効率を向上できます。さらに、Flink + Gateway + HiveSQL の互換性を活用して、OLAP システムを迅速に構築できます。Flink のもう一つの利点は、豊富なコネクタエコシステムを持ち、Flink の多彩なデータソースを活用した強力なフェデレーテッドクエリを実現できることです。たとえば、Hive データベースでのアドホッククエリだけでなく、Hive テーブルデータと MySQL、HBase、Iceberg、Hudi などのデータソース間のフェデレーテッドクエリも実行できます。

オフラインデータウェアハウスユーザーにとっては、Hive SQL の構文を使ってストリームコンピューティングジョブを記述できるようになり、リアルタイム化のコストを大幅に削減できます。従来の HiveSQL 構文をそのまま使用できますが、ストリーミングモードで実行されます。これを基盤として、ストリームバッチ統合 SQL レイヤーやストリームバッチ統合データベースレイヤーの構築もさらに探求できます。

2、Hive SQL 移行の課題

ただし、Flink による HiveSQL の移行サポートには多くの課題があり、主に以下の 3 つの側面に集約されます。

互換性：オフラインデータウェアハウス運用と Hive プラットフォームツールの互換性を含みます。主にアプリケーション層とプラットフォーム側の互換性に対応します。

安定性：移行後のジョブについて、まず本番環境での安定性を保証する必要があります。1.16 では、FLIP-168 の予測実行や Adaptive Hash Join など、多くの取り組みも行いました。この作業については、後日さらに詳しい記事を公開する予定です。

パフォーマンス：最後に、パフォーマンスも非常に重要です。1.16 では、動的パーティションプルーニング (DPP)、メタデータアクセスの高速化など、この分野でも多くの取り組みを行いました。この作業についても、後日さらに詳しい記事を公開する予定です。

次に、Hive 互換性に関連する作業にフォーカスします。

Hive 構文の互換性は、完全に新しい SQL エンジンを作成するのではなく、Flink SQL の多くのコアプロセスとコードを再利用しています。プラグ可能なパーサレイヤーを抽象化し、異なる構文のサポートと拡張に対応しています。Flink SQL は Flink パーサによって Flink RelNode に変換され、論理プランから物理プランに最適化され、最終的に Job Graph に変換されて実行されます。Hive 構文の互換性をサポートするため、Hive パーサコンポーネントを導入し、Hive SQL を Flink RelNode に変換しています。この過程で、Hive の既存の SQL 解析ロジックの大部分を再利用し、構文レベルの互換性を確保しています（すべて Calcite ベース）。その後、RelNode で同じプロセスとコードを再利用し、LogicalPlan、Physical Plan、JobGraph に変換してから、最終的に実行に送信されます。

アーキテクチャの観点から見ると、Hive 構文の互換性は複雑ではありませんが、「細部に悪魔が宿る」作業です。上記の図は、Flink 1.16 での Flink Hive 互換性に関する対応項目の一部を示しており、クエリ互換性、型システム、セマンティクス、動作、DDL、DML、補助クエリコマンドなど、多岐にわたる構文機能に対応しています。完了した対応項目の累計は約 100 件に達します。

Flink 1.16 では、Hive の互換性が 85% から 94.1% に向上しました。互換性テストは主に 12,000 件以上のテストケースを含む Hive qtest テストセットに依存しており、Hive の現在の主流構文機能をすべて網羅しています。互換性がない部分には ACID 機能（業界での使用は少ない）などが含まれます。ACID 機能を除外すると、互換性は 97% 以上に達します。

SQLGateway は Flink SQL のサーバーレイヤーコンポーネントです。独立したプロセスとして動作し、HiveServer2 コンポーネントに対抗する存在です。Flink の全体的なアーキテクチャにおいて、SQLGateway は中間に位置しています。

下位層では、Flink SQL と Hive SQL はともにユーザー API をカプセル化しています。Flink SQL と Hive SQL の両方が Flink のストリーミングおよびバッチランタイムを使用して実行され、バッチモードまたはストリームモードで実行できます。Flink のリソースは YARN、K8S、または Flink スタンドアロンクラスター上でもデプロイして実行できます。

上位層では、SQLGateway はプラグ可能なプロトコルレイヤーのエンドポイントを提供し、現在は HiveServer2 プロトコルと REST プロトコルの実装を用意しています。HiveServer2 エンドポイントを通じて、ユーザーは Hive エコシステムの多くのツールやコンポーネント（Zeppelin、Superset、Beeline、DBeaver など）を SQL Gateway に接続し、ストリームとバッチの統合 SQL サービスを提供しながら、Hive SQL との互換性も維持できます。REST プロトコルを通じては、Postman や curl コマンド、または Python や Java でのプログラミングでアクセスでき、完全で柔軟なストリームコンピューティングサービスを提供します。今後はエンドポイント機能をさらに拡張し、たとえばより高性能な gRPC プロトコルや PG プロトコルの互換性も提供する予定です。

3、Hive SQL の移行実践

現在、Kwai は Flink コミュニティと緊密に協力して、ストリームとバッチの統合を推進しています。Kwai では Hive SQL ジョブから Flink SQL ジョブへの移行が初期段階から進展しており、すでに数千のジョブが移行されています。Kwai の主な移行戦略は二重実行プラットフォームです。既存のサービスは継続して実行されます。二重実行プラットフォームにはインテリジェントなルーティングコンポーネントがあり、指定されたルールやパターンに基づいてジョブを識別し、MapReduce、Spark、または Flink に配信して実行します。初期段階では慎重に進め、一部のジョブをホワイトリスト機制で Flink での実行に指定し、安定性とパフォーマンスを観察して結果の整合性を比較してから、段階的にルールベースのスケーリングを行いました。より詳しい実践内容については、Flink Forward Asia 2022 で共有された「Practice of Hive SQL Migration to Flink SQL in Kwai」をご覧ください。

4、Hive SQL 移行のデモ

デモ 1：Hive SQL から Flink SQL への移行方法

次に、Hive SQL を Flink SQL に移行する方法を実演します。YARN クラスターと Hive 関連コンポーネント（HiveServer2 サービスを含む）を構築しています。データビジュアライゼーションと SQL クエリには Zeppelin を使用します。アドレスを 1 行変更するだけで Hive SQL から Flink SQL に移行できることを実演します。Zeppelin の操作感は同じで、SQL も修正不要です。完全なデモ動画については、完全なスピーチ動画をご覧ください：https://www.bilibili.com/video/BV1BV4y1T7d4

まず、Zeppelin で Hive インタープリターを設定し、HiveServer2 の JDBC アドレス、ポート、ユーザー名、パスワード、ドライバーなどの情報を入力します。

既存の Hive インタープリターで、Hive DDL コマンドを使って store_sale_detail ワイドテーブルを作成します。Hive SQL 構文を使って、store_sales、date_dim、store を関連付けてワイドテーブルに統合し、store_sale_detail に書き込みます。INSERT INTO ステートメントを実行すると、Hadoop プラットフォーム上で MapReduce タスクが実行されていることを確認できます。

store_sale_detail ワイドテーブルの作成が完了したら、クエリと分析を実行できます。たとえば、各店舗の日曜日の売上高を確認します。実行後、結果を円グラフなどの形式で表示できます。

上記のシンプルなデモでは、Hive を使ってデータの生成とデータ分析を行いました。計算エンジンは Hive ネイティブの Hadoop MapReduce ジョブで、YARN クラスター上で実行されています。次に、Flink SQL への移行を開始します。ジョブは引き続き YARN クラスター上で実行されます。

まず、Flink SQL クラスターをセットアップし、SQLGateway を起動します。Flink 1.16 をダウンロードして展開済みです。Hive コネクタ、JDBC コネクタ、MySQL ドライバは、lib フォルダの下に事前に用意してあります。さらに、opt ディレクトリの flink-table-planner-loader を flink-table-planner の JAR パッケージに置き換えてから、YARN セッションクラスターを起動します。セッションクラスターの起動後、YARN 上で Flink のセッションアプリケーションを確認できます。

SQLGateway を起動する前に、設定を変更する必要があります。主に HiveServer2 エンドポイントに関連する情報を設定します。

ここでは、SQLGateway のエンドポイントタイプを HiveServer2 にし、3 つの追加設定を行う必要があります：

HiveServer2 の hive-conf-dir、thrift.host、および thrift.port です。ポート番号は 20002 で起動していることに注意してください。その後、sql-gateway.sh start コマンドで SQL Gateway サービスを起動します。

起動後、移行を開始できます。HiveServer2 と同じマシン上で実行されているため、ポート番号を変更するだけで済みます。ここの 10000 ポートを先ほど起動した 20002 ポート（Flink SQLGateway のポート）に変更します。その他の変更は不要です。インタープリターを再起動すれば、移行は完了です。

その後、Zeppelin で Hive SQL ステートメントを再実行すると、結果が一致していることを確認できます。

上記の図に示すのは、各店舗の日曜日の売上高合計を照会した結果です。円グラフの結果は Hive エンジンの照会結果と完全に一致しています。異なるのは、このクエリが Flink エンジン上で実行されている点です。

Hive SQL から Flink SQL に移行すると、パフォーマンスの向上だけでなく、Flink SQL が提供する追加の機能も活用できます。より豊富なフェデレーテッドクエリやストリームバッチ処理機能が含まれます。

Flink DDL を使って新しいカタログを作成できます。たとえば、MySQL テーブルに Hive にはない新しいディメンション情報があり、それを関連付けて新しいデータ探索を行いたい場合、Flink の CREATE CATALOG ステートメントで MySQL カタログを作成し、フェデレーテッドクエリを実現できます。同時に、Flink は MySQL 側にプッシュダウン可能なプロジェクトとフィルターをプッシュダウンして最適化します。

さらに、Hive SQL でストリームコンピューティングの機能も体験できます。Flink 構文で datagen テーブルを作成すると、ランダムデータが継続的に生成されます。Hive 構文に戻して Hive 結果テーブルシンクを作成します。実行モードをストリーミングに切り替え、INSERT INTO ステートメントを実行すると、ストリームジョブが投入され、datagen で生成されたデータが Hive に継続的に書き込まれます。

Hive 結果テーブルにストリームジョブによってデータが継続的に書き込まれていることを確認するため、Hive 構文で書き込まれたテーブルをクエリすることもできます。上記の図に示すように、count (*) ステートメントを連続して実行すると、テーブルにデータが継続的に書き込まれているため、クエリ結果が常に変化することが確認できます。

5、今後の計画

今後、Flink は以下の 3 つの側面で継続的に進化していきます。

第一に、バッチコンピューティングでの取り組みと投資を継続し、バッチの安定性とパフォーマンスを向上させ、短期的に主流のバッチ計算エンジンに追いつくことを目標とします。

第二に、データレイク分析の改善です。より効率的なバッチデータレイクの読み書き、クエリ最適化のプッシュダウン、カラムナーストレージでの読み書き最適化、Iceberg、Hudi、Flink Table Store のサポートなどが含まれます。さらに、スナップショットバージョンの照会、メタデータの照会、UPDATE、DELETE、MERGE INTO などの豊富な DML 構文、CALL コマンドによるデータレイクデータの管理機能など、豊富なレイクデータクエリと管理機能も提供する予定です。

第三に、Flink バッチエコシステムの構築です。Remote Shuffle Service とデータリネージ管理のさらなる改善を含みます。

Q&A

Q：Hive の書き込みを Flink で実行する場合、Hive に大量のデータがあると、メモリ不足や OOM などのエラーが発生しますか？

A：現在、Flink のバッチモードでは、ほぼすべてのオペレーターにメモリ管理機構が備わっています。データは Java オブジェクトとして Flink に格納されるのではなく、Java メモリ内に独立したメモリ領域が確保されて使用されます。メモリがいっぱいになった場合は、ディスクへの退避とスピリングが行われます。速度は若干低下する可能性がありますが、通常メモリ OOM は発生しません。

Q：Flink は Hive のカスタム UDF 関数をサポートしていますか？移行コストはどの程度ですか？

A：はい、直接移行できます。

Q：既存のオフラインデータウェアハウスを Hive から Flink に移行する際のリスクはありますか？スムーズな移行のための注意点を教えてください。

A：現在、スムーズな移行には主に二重実行プラットフォームを使用しています。一部のジョブを機制を通じて選択して移行し、移行後のジョブは両プラットフォームで同時に実行されます。そのため、動作と結果の整合性を検証し、その後段階的に旧プラットフォームのジョブを停止して単一実行に移行する必要があります。プロセス全体を段階的に進める必要があり、通常は半年から 1 年程度かかります。

Q：デモの SQL クエリでは Hive on MR エンジンが使用されていました。移行後は Flink SQLGateway を使用しますか、それとも Hive on MR モードを使用しますか？

A：移行後は、設定されたポートが Flink SQL Gateway のポートであるため、SQL リクエストは Flink SQL Gateway を経由します。Gateway が Hive SQL を Flink ジョブにコンパイルし、YARN クラスターに投入して実行します。

Q：Flink がバッチタスクを実行する際、TaskManager の数はユーザーが指定しますか、それとも自動的に生成されますか？

A：スタンドアロンモード（K8S 上で実行されるスタンドアロンモードを含む）では、TaskManager の数はユーザーが指定します。その他のモード（YARN/K8S アプリケーションモード、YARN セッションモード、YARN パージョブモード、K8S ネイティブセッションモードなど）では、Flink が TaskManager の数を決定して起動します。起動される TaskManager の数は、ジョブが要求するスロット数に関連します。taskmanager.numberOfTaskSlots パラメーターがスロット数と TaskManager 数のマッピング関係を決定します。スロット数は、スケジュールされたジョブノードの同時実行数に関連します。

Q：Flink が K8S 上で実行され、動的リソース割り当てが有効な場合、シャッフルデータは常に Pod ディスクに保存されますか？

A：TaskManager 上または RemoteShuffleService 上のいずれかを選択できます。

Q：オフラインジョブの移行後、WITH AS 構文と PARTITION BY 構文は引き続きサポートされますか？

A：WITH AS 構文は引き続きサポートされ、CREATE TABLE の PARTITIONED BY 構文もサポートされます。