An Enterprise-level Distributed Batch Processing Solution Based on Task Scheduling

背景

まずは分散バッチ処理について説明します。文字通り、アプリケーションによるバッチ計算および処理を必要とする大量のビジネスデータが存在します。しかし、スタンドアロンモードでの実行では処理時間が長くなる可能性があり、ビジネスクラスター内の各アプリケーションノードの処理能力を十分に活用できません。一般的な分散バッチ処理方式は、ビジネスクラスター内のすべてのビジネスアプリケーションノードが協調して大量のデータ処理タスクを完了させることで、全体的な処理効率と信頼性を効果的に向上させることができます。

バッチモデル

シンプルなスタンドアロンシナリオでは、複数のスレッドを有効化して大規模なタスクを同時に処理できます。複数マシン環境では、複数のマシンが同じタスクを並列で同時に処理できます。したがって、分散バッチ処理方式は、タスク分割、配布、並列実行、結果集約、フェイルオーバー、動的拡張などのビジネスアプリケーションクラスター間の分散協調ロジックをコード開発レベルで開発者から隠蔽し、ユーザーが前述のビジネスロジック分割ルールとビジネスロジック処理のみに集中できるようにする必要があります。

ビッグデータバッチ処理との比較

ビッグデータ処理シナリオでは MapReduce モデルも使用されますが、その処理ロジックの本質はここで議論するビジネスバッチ処理ロジックと一致しています。ビッグデータシナリオでは、バッチ処理は主にデータ自体を対象としており、データストレージとデータバッチ処理をサポートする対応するビッグデータプラットフォームクラスターのデプロイが必要です。したがって、このシナリオの主な目的は完全なデータプラットフォームを構築することです。ビッグデータバッチ処理シナリオと比較して、今回は主に分散型ビジネスバッチ処理シナリオに焦点を当て、既存のビジネスアプリケーションサービスクラスターに基づいて分散バッチ処理ロジックを構築します。以下の要件は分散バッチ処理ソリューションで対応できます。

- 時間のかかるビジネスロジックをデカップリングし、コアリンクビジネス処理への高速応答を確保する

- ビジネスクラスターのすべてのアプリケーションノードを完全にスケジュールして協調させ、ビジネス処理をバッチで完了する

- ビッグデータ処理とは異なり、サブタスク処理中に他のオンラインビジネスサービスもバッチ処理プロセスに参加するために呼び出される

オープンソースバッチ処理方式

ElasticJob

ElasticJob は分散型タスクスケジューリングフレームワークです。主な特徴は、Quartz に基づいたスケジュール機能を実装し、ビジネスクラスター内でタスクの分割と協調処理を行う機能を提供することです。アーキテクチャ全体は ZooKeeper に基づいており、タスクシャーディング実行、アプリケーションクラスターの動的弾性スケジューリング、サブタスク実行の高可用性を実現します。シャーディングモデルにより、大量のビジネスデータ処理をビジネスクラスター内の各ノードに均等に分散して処理でき、タスク処理効率を効果的に向上させます。

- SimpleJob

Spring Boot プロジェクトでは、YAML を通じてタスク定義を設定でき、タスク実装クラス、スケジュール周期、シャーディング情報を指定します。

設定した org.example.job.SpringBootSimpleJob クラスは、SimpleJob インターフェイスの execute メソッドを実装し、ShardingContext パラメーターを通じて対応するビジネスフラグメントデータを取得してビジネスロジック処理を行う必要があります。

3 つのアプリケーションサービスをスケジュール処理クラスターとしてデプロイし、上記のタスクを処理します。タスクがトリガーされて実行されると、ElasticJob は対応する 3 つのシャーディングタスクを 3 つのアプリケーションサービスに割り当て、タスクデータ処理全体を完了させます。

- DataflowJob

現在のところ、DataflowJob と SimpleJob の全体的な構造に本質的な違いはありません。以下のインターフェイスを参照すると、SimpleJob と比較して、ビジネス側が処理対象データを自律的に読み込むための fetchData メソッドが追加されています。事実上、SimpleJob の execute メソッドを論理的に 2 つのステップに分割したものです。唯一の違いは、DataflowJob は常駐型データ処理タスク（ストリーミングプロセスと呼べる）を提供し、fetchData が空になるまでタスクを永続的に実行できる点です。

DataflowJob タスクの YAML 設定に `streaming.process=true` を追加することで、このタスクに対してストリーミングプロセスの効果を有効化できます。タスクがトリガーされて実行されると、各シャーディングタスクは対応するプロセスに従ってループ実行されます。fetchData -> processData -> fetchData の順で、fetchData が空になるまで続きます。このモードのシナリオ分析は以下の通りです。

- 単一のシャーディングタスクで待機データ量が多い場合。データ取得時に、シャーディングの部分的なページングデータを読み取り、すべてのデータが処理されるまで処理を続ける

- 断片化された保留データが継続的に生成される場合。fetchData を通じてデータを継続的に取得でき、長期的な常駐と継続的なビジネスデータ処理を実現する

- 特徴分析

ElasticJob の分散型シャーディングモデルは、一般的でシンプルなバッチ処理シナリオに大きな利便性とサポートを提供し、大量のビジネスデータ処理における分散型シャーディング実行の全協調プロセスを解決します。一方で、以下の側面においていくつかの課題が残る可能性があります。

- アーキテクチャ全体のコアが ZK の安定性に依存している

- 追加の運用保守と高可用性の確保が必要

- 大量のタスクが ZK に保存されてトリガーされ、実行プロセスも ZK に依存する。タスク量が多い場合、ZK クラスターがスケジューリングのパフォーマンスボトルネックになりやすい

- シャーディング設定数が固定で、動的シャーディングがサポートされていない

- 各パーティションの処理データ量に大きな差がある場合、クラスターの処理能力バランスが崩れやすい

- シャーディング定義が不合理な場合、クラスター規模がシャーディング数より大幅に大きいとクラスターの弾力性が効果を失う

- シャーディング定義とビジネスロジックが分断されやすく、両者の関係を手動で維持することが困難

- 管理コンソール機能が弱い

Spring Batch バッチ処理フレームワーク

Spring Batch バッチ処理フレームワークは、軽量で洗練されたバッチ処理機能を提供します。Spring Batch のタスクバッチ処理では、主にシングルプロセスマルチスレッドと分散型マルチプロセスの 2 つの方法を提供します。シングルプロセスマルチスレッド処理モードでは、ユーザーはカスタムジョブをバッチタスクユニットとして定義できます。ジョブは直列または並列の 1 つ以上のステップで構成されます。各ステップはリーダー、プロセッサー、ライターで構成され、タスクの各ステップの読み取り、処理、出力を完了します。以下の議論では、ジョブが 1 つのステップのみを含むシナリオの分析に焦点を当てます。

個人的には、シングルプロセス内のマルチスレッドに関して、Spring Batch フレームワークの実用的な意義はあまり大きくないと考えます。主な理由は、小規模なデータタスクを処理するためにこのフレームワークを導入するのはやや手間がかかることです。スレッドプールを独自に構築して解決することも十分可能です。ここでは、一定規模のビジネスクラスターにおける分散型協調によるビジネスデータバッチ処理タスクのシナリオに主に焦点を当てます。Spring Batch はリモートチャンキングおよびリモートパーティショニング機能を提供します。ジョブのステップ内で、タスクを特定のルールに従って複数のサブタスクに分割し、クラスター内の他のワーカーに配布して処理することで、分散型並列バッチ処理を実現します。リモートインタラクション機能は、通常、サードパーティのメッセージミドルウェアに依存してサブタスクの配布と実行結果の集約を実現します。

- リモートチャンキング

リモートチャンキングは、Spring Batch が大量のデータタスクを処理する際に提供する分散型バッチ処理ソリューションです。単一のステップ内で ItemReader を通じてデータを読み込み、複数のチャンクを構築できます。ItemWriter はこれらのチャンクをメッセージミドルウェアまたはその他の形式でクラスターノードに配布します。クラスターアプリケーションノードは各チャンクに対してビジネス処理を実行します。

リモートチャンキングの例

上記のメインノードでは、ItemReader と ItemWriter は本稿で議論するバッチモデルの「タスク分割」フェーズに対応付けられます。メインノードの ItemWriter には、Spring Batch Integration が提供する ChunkMessageChannelItemWriter を使用でき、Spring Integration が提供する他のチャネル（AMQP や JMS など）との統合を通じてバッチタスクデータの読み込みとチャンク配布を完了します。

スレーブノードは主に、配布されたチャンクデータ（サブタスクと理解できる）に対して対応するビジネスロジック処理とデータ結果出力を実行します。したがって、サブタスク処理側では、Spring Batch Integration が提供する ChunkProcessorChunkHandler を設定し、サブタスクの受信、実際のビジネス処理、処理結果のフィードバックなどの関連アクションを完了する必要があります。

- リモートパーティショニング

リモートパーティショニングとリモートチャンキングの主な違いは、マスターノードがデータ読み込みを担当しない点です。Partitioner を通じて現在のステップを複数のサブステップ（サブタスクとも呼ばれる）に分割し、PartitionHandler を通じて対応するサブタスクを各スレーブノードに配布して処理すると理解できます。Spring Batch Integration は MessageChannelPartitionHandler を提供して対応するサブタスク配布を実現します。基盤となるレイヤーでは、メッセージミドルウェアへの依存による適応と接続も必要です。各スレーブノードでは、サブタスクのステップのコンテキスト情報を読み取り、その情報に基づいて完全な ItemReader、ItemProcess、ItemWrite 処理を実行する必要があります。

- 特徴分析

Spring Batch フレームワークの包括的な特徴分析は以下の通りです。

- 完全なバッチ処理機能：シングルマシンマルチスレッド、分散型マルチプロセス協調バッチ処理をサポートし、カスタマイズ可能なシャーディングモデルに対応。

- スケジュール機能の未提供：ネイティブのスケジュール機能にはサードパーティのタイミングフレームワークの統合が必要（例：Spring Task ではクラスター内の重複トリガーを独自に解決する必要がある）。

- 視覚的管理機能が弱い：Spring Batch は一般的にプログラムやファイルでタスクを設定し、管理コンソールは別途構築する必要があり、管理機能が弱い。

統合の難易度が高い：分散型バッチ処理機能には追加のサードパーティミドルウェアの統合、またはインターフェイスに基づく独自開発が必要。公式提供の方法に基づくエンタープライズレベルの使用には、比較的複雑な計画と統合が伴う。

エンタープライズレベルのバッチ処理方式 - SchedulerX 視覚的 MapReduce タスク

SchedulerX タスクスケジューリングプラットフォームは、エンタープライズレベルのバッチ処理ニーズに対して完全な総合ソリューションを提供します。ユーザーはパブリッククラウドプラットフォームのサービスを直接使用することで、ビジネスアプリケーションクラスターの分散型バッチ処理機能を簡単に実現できます（非 Alibaba Cloud のビジネスアプリケーションデプロイにも対応可能）。追加のミドルウェア統合メンテナンスのためのデプロイは不要です。

原理分析

ソリューション全体において、タスクスケジューリングプラットフォームはユーザーに包括的な視覚的管理、信頼性の高いスケジュール実行、登録タスクの視覚的クエリ機能を提供します。さらに、ユーザービジネスアプリケーション側に SchedulerX SDK を統合することで、分散型バッチ処理機能への迅速なアクセスを実現できます。このとき、ユーザーはバッチ処理モデルにおけるサブタスクのビジネス分割ルールと各サブタスクの処理ロジックのみに集中すればよくなります。この分散型バッチ処理には以下の特徴があります。

- サブタスクの高可用性：クラスター実行ノードのダウン時に自動フェイルオーバーをサポートし、オフラインマシンのサブタスクを他のノードに再配布

自動弾性拡張：クラスターに新しいアプリケーションノードがデプロイされると、自動的に後続タスクの実行に参加

- 可視化機能：タスクおよびサブタスクの実行プロセスに対する多様なモニタリング、運用保守、ビジネスログクエリ機能を提供

以下に一般的な原理プロセスを説明します。

- プラットフォームで MapReduce タスクが作成されると、スケジュールサービスが信頼性の高いスケジュールトリガー実行を有効化

- MapReduce タスクの実行がトリガーされると、スケジュールサービスはアクセスされたビジネスワーカーノードの中から 1 つのノードをこのタスク実行のメインノードとして選択

- メインノードはユーザー定義開発のサブタスク分割および読み込みロジックを実行し、map メソッド呼び出しを通じてサブタスク処理リクエストをクラスター内の他のワーカーノードに均等に配布

- メインノードは分散バッチ処理タスク全体の処理プロセスと各ワーカーノードのヘルスモニタリングを行い、全体的な運用と高可用性を確保

- 他のワーカーノードはサブタスク処理リクエストを受信後、コールバックしてユーザー定義のビジネスロジックを実行し、各サブタスクの処理要件を最終的に完了。単一アプリケーションノードで同時に処理できるサブタスクの並列スレッド数は設定可能

- すべてのサブタスクが完了した後、メインノードはすべてのサブタスクの実行結果を集約して reduce メソッドをコールバックし、スケジューリングプラットフォームにフィードバックして今回の実行結果を記録

開発者はビジネスアプリケーション内で MapReduceJobProcessor 抽象クラスを実装し、isRootTask で今回処理対象のビジネスサブタスクデータオブジェクトリストを読み込むだけです。非ルートリクエストでは、jobContext.getTask() を通じて単一サブタスクオブジェクト情報を取得し、それに基づいてビジネス処理ロジックを実行します。ビジネスアプリケーションのデプロイがクラスターノードに公開された後、タスクがトリガーされて実行されると、クラスターのすべてのノードが参加して全体の分散バッチ処理タスクを完了まで協調実行します。

機能上の利点

- サブタスク可視化機能

ユーザーダッシュボード：すべてのタスクのトリガーおよび実行記録情報を視覚的に表示します。

サブタスク詳細の可視化：タスク実行記録の詳細を照会することで、各サブタスクの実行ステータスとノード情報を取得できます。

- サブタスクビジネスログ

サブタスク一覧で「ログ」をクリックすると、現在のサブタスク処理中のログ情報を取得できます。

- 実行スタックビュー

実行スタックビュー機能により、サブタスク処理が停止して操作が完了しないシナリオにおいて、対応する実行スレッドのスタック情報を簡単にトラブルシューティングできます。

- カスタムビジネスタグ

サブタスクビジネスタグ機能は、ユーザーにサブタスクビジネス情報を迅速かつ視覚的に確認・照会する機能を提供します。下図の「アカウント名」は、このサブタスクから分割されたビジネスタグ情報です。ユーザーはこの情報に基づいて対応するビジネスサブタスクの処理ステータスを迅速に把握でき、指定したビジネスタグ情報のサブタスク処理ステータスを照会できます。

サブタスクにカスタムラベルを設定するには、この map で配布されるサブタスクオブジェクトに対して BizSubTask インターフェイスを実装し、その labelMap メソッドを実装するだけで、各サブタスクに独自のビジネス特徴ラベルを追加して視覚的クエリに利用できます。

- オープンソース互換

SchedulerX は XXL-Job や ElasticJob を含む、一般的なオープンソースフレームワークに基づいて記述されたエグゼキューターをサポートしています。今後のスケジューリングプラットフォームでは Spring Batch タスクのスケジューリングもサポートする予定です。

ケースシナリオ

分散型バッチ処理モデル（視覚的 MapReduce モデル）は、実際のエンタープライズレベルのアプリケーションで多数の需要シナリオに存在します。一般的な使用シナリオは以下の通りです。

- データベースまたはテーブル分割データのバッチ並列処理。データベースまたはテーブル分割情報をクラスターノード間に配布して並列処理を実現

- 都市・地域別の物流注文データ処理。都市と地域をサブタスクオブジェクトとしてクラスターノード間に配布して並列処理を実現

- 視覚的 MapReduce サブタスク可視化機能を活用し、重要顧客や注文情報をサブタスク処理オブジェクトとして対応するデータレポート処理や情報プッシュを行い、重要サブタスクの視覚的追跡を実現

- ファンド販売ビジネスケース

以下にファンド販売のビジネスケースを参考に提供します。分散型バッチ処理モデルを使用すれば、ユーザーは独自のビジネスシナリオで自由に活用できます。ケース説明：毎日、投資家の口座および取引アプリケーションデータがファンド会社とファンド販売会社（Ant Wealth など）の間で同期的に処理され、通常はファイルデータによるやり取りが行われます。ファンド会社は N 社以上のベンダーと完全に独立しており（逆も同様）、各ベンダーが提供するデータファイルも完全に独立しています。各ベンダーのデータファイルには、ファイル検証、インターフェイスファイル解析、データ検証、データインポートなどの複数の固定ステップが必要です。上記の固定ステップを処理する際、ファンド会社は分散バッチ処理を採用してデータファイル処理の高速化に適しています。各ベンダーをサブタスクオブジェクトとしてクラスターに配布し、すべてのアプリケーションノードが割り当てられた異なるベンダーのデータファイル解析処理に参加します。

このケースの主な目的は、ファンド取引クリアリングのビジネスステップを並列バッチ処理で処理することです。後続の各処理ステップも同様の方法で処理できます。さらに、各視覚的 MapReduce タスクノードは DAG 依存関係のオーケストレーションを通じて、完全な自動ビジネスクリアリングプロセスを構築できます。

まとめ

分散型タスクスケジューリングプラットフォーム SchedulerX は、エンタープライズレベルの分散バッチ処理に対する包括的なソリューションを提供し、ユーザーに迅速で使いやすいアクセスモードを提供するとともに、スケジュール実行、視覚的な運用追跡、管理可能でシンプルな運用保守、および高可用性のスケジューリングサービスをサポートします。同時に、大規模ディスク、ログサービス、監視アラートなどのエンタープライズレベルのモニタリング機能を備えています。