Application of Serverless Asynchronous Task Processing System in the Field of Data Analysis

非同期タスク処理システムにおけるデータ分析

データ処理、機械学習トレーニング、およびデータ統計分析は、オフラインタスクの最も一般的なタイプです。このタイプのタスクは、一連の前処理を経た後、アップストリームからタスクプラットフォームに一括送信され、バッチトレーニングと分析が行われます。処理言語の面では、Python が豊富なデータ処理ライブラリを備えていることから、データ分野で最もよく使用される言語の一つとなっています。関数コンピューティングは Python ランタイムをネイティブでサポートし、サードパーティライブラリの迅速な導入にも対応しているため、関数コンピューティングを非同期タスクに使用するのは非常に便利です。

データ分析シナリオの一般的な要件

データ分析シナリオは、実行時間が長く、同時実行数が多いという特徴があります。オフラインシナリオでは、大量のデータが定期的にトリガーされ、集中的に処理されることがよくあります。このトリガー特性のため、ビジネス側はリソース利用率 (コスト) に対する要件が高く、効率を満たしながらコストを最小限に抑えることを望んでいます。具体的な要件は以下のとおりです。

1. プログラム開発が容易で、サードパーティパッケージやカスタム依存関係への対応が柔軟であること。

2. 長時間実行に対応し、実行中にタスクのステータスを確認したり、マシンにログインして操作を実行したりできること。データエラーが発生した場合はタスクを手動で停止できること。

3. リソース利用率が高く、コストが最適であること。

上記の要件は、関数コンピューティングの非同期タスクを使用するのに非常に適しています。

典型的な事例 - Database Autonomy Service

基本ビジネス情報

Alibaba Cloud Group 内のデータベースパトロールプラットフォームは、主に SQL ステートメントのスロークエリとログの最適化および分析に使用されています。プラットフォーム全体のタスクは、オフライン トレーニングとオンライン分析という 2 つの主要タスクに分かれています。オンライン分析ビジネスのコンピューティング規模は数万コアに達し、オフラインビジネスの 1 日あたりの実行時間も数万コア時間に上ります。オンライン分析とオフライン トレーニングのタイミングの不確実性により、クラスター全体のリソース利用率を向上させることが難しく、ビジネスピーク時には強力なエラスティックコンピューティング能力のサポートが必要です。関数コンピューティングの導入後、ビジネス全体のアーキテクチャを以下に示します。

ビジネス上の課題とアーキテクチャの進化

データベースパトロールプラットフォームは、Alibaba 全ネットワークの各リージョンにおけるデータベース SQL の最適化と分析を担当しています。MySQL データは各リージョンのさまざまなクラスターから収集され、リージョン単位で事前集約されて一元的に保存されます。分析時にはリージョン間の集約と統計処理が必要なため、パトロールプラットフォームはまずイントラネット上に大規模な Flink クラスターを構築して統計分析を試みましたが、実際の運用で以下の問題が生じました。

1. データ処理アルゴリズムの反復が煩雑であること。主にアルゴリズムのデプロイ、テスト、リリースに反映されており、Flink のランタイム機能がリリースサイクルを大幅に制限しています。

Flink は一般的なサードパーティライブラリやカスタマイズされたサードパーティライブラリのサポートが十分ではありません。アルゴリズムが依存する機械学習ライブラリや統計ライブラリの一部は、Flink の公式 Python ランタイムで提供されていないか、バージョンが古く、使いにくく、要件を満たせません。

3. Flink の転送リンクが長く、Flink のトラブルシューティングが困難です。

4. ピーク時には、弾力性とリソースの両面で要件を満たすことが困難です。さらに全体のコストが非常に高くなります。

関数コンピューティングを理解した後、Flink 計算部分のアルゴリズムタスクの移行を行い、コアトレーニングおよび統計アルゴリズムを関数コンピューティングに移行しました。関数コンピューティングの非同期タスクが提供する関連機能を活用することで、開発、運用保守、およびコストの面で大幅に改善されました。

関数コンピューティングアーキテクチャ移行の効果

関数コンピューティングへの移行後、システムはピークトラフィックに完全に対応でき、日々の分析およびトレーニングタスクを迅速に完了できます。

2. 関数コンピューティングの豊富なランタイム機能により、ビジネスの迅速な反復がサポートされます。

3. 同じコア数でのコストは、元の Flink の 1/3 に計算されます。

関数コンピューティングの非同期タスクは、このようなデータ処理タスクに非常に適しています。関数コンピューティングは、プラットフォーム運用保守の複雑さから解放されながらコンピューティングリソースのコストを削減し、アルゴリズムの開発と最適化に集中できる環境を実現します。

関数コンピューティングの非同期タスクのベストプラクティス - Kafka ETL

ETL はデータ処理において比較的一般的なタスクです。元データは Kafka または DB に存在し、ビジネス上のデータ処理を経て、他のストレージメディアにダンプ (または元のタスクキューに保存) する必要があります。このタイプのビジネスも明確なタスクシナリオに属します。

クラウド上のミドルウェアサービス (クラウド上の Kafka など) を使用する場合、関数コンピューティングの強力なトリガーを活用することで、Kafka Connector のデプロイやエラー処理といったビジネス関連の運用に悩まされることなく、Kafka との統合を容易に実現できます。

ETL タスクシナリオの要件

ETL タスクは通常、ソース、シンク、および処理ユニットの 3 つの部分で構成されます。したがって、計算能力の要件に加えて、ETL タスクにはタスクシステムの強力なアップストリーム・ダウンストリーム連携エコシステムが必要です。さらに、データ処理の正確性の要件から、タスク処理システムがタスクの重複排除および Exactly Once の運用セマンティクスを提供できる必要があります。また、処理失敗メッセージへの補償処理能力 (リトライやデッドレターキューなど) も求められます。まとめると以下のとおりです。

1. タスクの正確な実行：

A. タスクの重複トリガー時に重複排除をサポートすること。

B. タスクの補償処理、デッドレターキューをサポートすること。

2. タスクのアップストリームとダウンストリーム：

A. 簡単にデータを取得し、処理後に他のシステムに転送できること。

オペレータ能力の要件：

A. カスタムオペレータをサポートし、柔軟にさまざまなデータ処理タスクを実行できること。

ETL タスクに対する Serverless Task のサポート

関数コンピューティングがサポートする転送先設定機能は、アップストリーム・ダウンストリームの便利な連携とタスクの正確な実行という ETL タスクの要件を十分にサポートできます。関数コンピューティングの豊富なランタイムサポートにより、データ処理タスクも非常に柔軟になります。Kafka ETL タスク処理シナリオで使用する主なサーバーレスタスク機能は以下のとおりです。

非同期ターゲット設定機能：

A. タスク成功時のターゲットを設定することで、ダウンストリームシステム (キューなど) へのタスクの自動配信をサポートします。

B. タスク失敗時のターゲットを設定することで、デッドレターキュー機能をサポートし、失敗したタスクをメッセージキューに配信して後続の補償処理を待機します。

柔軟なオペレータとサードパーティライブラリのサポート：

Python は、統計や計算に関するサードパーティライブラリの豊富なサポートにより、データ処理分野で最も広く使用されている言語の一つです。関数コンピューティングの Python ランタイムはサードパーティライブラリのパッケージングをサポートし、迅速なプロトタイプ検証とテスト実行を可能にします。

Kafka ETL タスク処理例

シンプルな ETL タスク処理を例に取ります。データソースは Kafka です。

関数コンピューティングでの処理後、タスク実行結果とアップストリーム・ダウンストリーム情報がメッセージサービス MNS にプッシュされます。プロジェクトの関数コンピューティング部分のソースコードを以下に示します。

https://github.com/awesome-fc/Stateful-Async-Invocation

リソース準備

Kafka リソース準備

Kafka コンソールに入り、[インスタンスの購入] をクリックしてデプロイします。インスタンスのデプロイが完了するまで待機します。

作成したインスタンスに入り、テスト用の Topic を作成します。

ターゲットリソース準備 (MNS)

MNS コンソールに入り、2 つのキューを作成します。

1. デッドレターキュー：デッドレターキューとして使用します。メッセージ処理が失敗した場合、実行コンテキスト情報がここに送信されます。

2. fc etl processed message：タスク正常実行後のプッシュターゲットとして使用します。

作成後、以下の図のようになります。

デプロイ

1. Serverless Devs をダウンロードしてインストールします。

npm install @serverless-devs/s

詳細なドキュメントについては、Serverless Devs インストールドキュメントを参照してください。

2. キー情報を設定します。

s config add

詳細なドキュメントについては、Alibaba Cloud キー設定ドキュメントを参照してください。

3. プロジェクトに入り、s.yaml ファイル内のターゲット ARN を上記で作成した MNS キューの ARN に変更し、サービスロールを既存のロールに変更します。

4. デプロイ：s deploy - t s.yaml

ETL タスクの設定

Kafka コンソールのコネクタタスクリストタブに入り、[コネクタの作成] をクリックします。

基本情報とソース Topic を設定した後、ターゲットサービスを設定します。ここでは関数コンピューティングをターゲットとして選択します。

ビジネス要件に基づいて、送信バッチサイズとリトライ回数を設定できます。これでタスクの基本設定が完了しました。注意：送信モードは「非同期」を選択してください。

関数コンピューティングの非同期設定ページに入ると、現在の設定は以下のようになっています。

ETL タスクのテスト

Kafka コンソールのコネクタタスクリストタブに入り、[テスト] をクリックします。メッセージ内容を入力した後、[送信] をクリックします。

複数のメッセージを送信した後、関数コンソールに入ります。複数のメッセージが実行中であることが確認できます。ここで、タスクを停止する方法を使用してタスク実行の失敗をシミュレートします。

メッセージサービス MNS コンソールに入ると、以下を確認できます。

キューの詳細に入ると、2 つのメッセージ内容を確認できます。成功メッセージの内容を例に取ります。

ここでは、「responsePayload」キーに関数から返された元のコンテンツを確認できます。一般的には、データ処理の結果をレスポンスとして返すため、後続の処理で「responsePayload」を読み取ることで処理済みの結果を取得できます。

「requestPayload」キーは Kafka トリガー関数で計算された元のメッセージ内容です。このデータの内容を読み取ることで、元のデータを取得できます。

関数コンピューティングの非同期タスクのベストプラクティス - オーディオおよびビデオ処理

コンピュータ技術とネットワークの発展に伴い、ビデオオンデマンド技術は、優れたユーザーインタラクションとストリーミングメディア配信技術により、教育やエンターテインメントなどの業界で高く評価されています。現在、クラウドコンピューティングプラットフォームメーカーの製品ラインは絶えず成熟・完善しており、ビデオオンデマンドアプリケーションを構築する場合、クラウドを直接利用することでハードウェア調達や技術面などのさまざまな障壁を解消できます。Alibaba Cloud を例に取ると、典型的なソリューションは以下のとおりです。

このソリューションでは、OSS が大量の動画の保存をサポートし、収集・アップロードされた動画はトランスコーディングされてさまざまな端末に対応し、CDN により端末デバイスでの動画再生速度が高速化されます。さらに、ポルノやテロリズムの検出など、コンテンツセキュリティレビューの要件もあります。

オーディオとビデオ処理は典型的な長時間処理シナリオであり、関数コンピューティングタスクの活用に非常に適しています。

オーディオおよびビデオ処理の要件

ビデオオンデマンドソリューションにおいて、動画トランスコーディングは最も計算集約的なサブシステムです。クラウド上の専用トランスコーディングサービスを使用できますが、以下のシナリオでは独自のトランスコーディングサービスを構築することを選択します。

- より柔軟な動画処理サービスが必要な場合。たとえば、仮想マシンまたはコンテナプラットフォーム上で FFmpeg ベースの動画処理サービスをデプロイ済みだが、それを基にリソース利用率を改善し、ピークとバレーが顕著でトラフィックが急増する場合にも迅速な弾力性と安定性を実現したい場合。

- 複数の大容量動画を迅速にバッチ処理する必要がある場合。たとえば、毎週金曜日に 4 GB を超える 1080P の大規模動画が定期的に数百件生成され、各タスクの実行に数時間かかる場合。

- 動画処理タスクの進捗をリアルタイムで把握したい場合。また、エラーが発生した場合にインスタンスにログインしてトラブルシューティングを行ったり、タスクの実行を停止してリソース消費を回避したりする必要がある場合。

オーディオおよびビデオシナリオに対する Serverless Task のサポート

上記の要件は典型的なタスクシナリオです。このようなタスクはピークとバレーの特性を持つことが多いため、コンピューティングリソースの運用保守をどのように行い、コストを最小限に抑えるかは、実際の動画処理ビジネス自体のワークロードを上回ります。

Serverless Task の製品形態は、こうしたシナリオに対応するために生まれました。Serverless Task を使用することで、高い弾力性、高い可用性、低コスト、かつメンテナンスフリーな動画処理プラットフォームを迅速に構築できます。

このシナリオで使用する Serverless Task の主な機能は以下のとおりです。

1. 無料の運用保守と低コスト：コンピューティングリソースはオンデマンドで使用でき、使用しない場合は料金が発生しません。

2. 長期的な実行タスク負荷への対応：単一のインスタンスで最大 24 時間の実行をサポートします。

3. タスクの重複排除：トリガー側でのエラー補償をサポートします。単一タスクに対して、Serverless Task は自動的に重複排除を行う機能を備えており、より信頼性の高い実行を実現します。

4. タスクの可観測性：実行中、実行成功、および実行失敗のすべてのタスクを追跡およびクエリできます。タスク実行履歴データのクエリおよびタスクログクエリをサポートします。

5. タスクの操作性：タスクの停止とリトライが可能です。

6. アジャイル開発とテスト：S ツールを使用した自動ワンクリックデプロイを公式サポートします。実行中の関数インスタンスへのログイン機能をサポートします。インスタンスに直接ログインして ffmpeg などのサードパーティプログラムをデバッグでき、見たものがそのまま得られます。

Serverless - FFmpeg 動画トランスコーディング

初期化プロジェクト：s init video transcode - d video transcode

プロジェクトに入り、デプロイします：cd video transcode&&s deploy

関数の呼び出し

5 つの非同期タスク関数呼び出しを開始します

FC コンソールにログインします

各トランスコーディングタスクの実行状況を明確に確認できます。

- A の動画トランスコーディングがいつ開始され、いつ終了したか

- B の動画トランスコーディングタスクが予期しない結果の場合、途中で呼び出しを停止できます

- 呼び出し状態フィルタリングとタイムウィンドウフィルタリングにより、現在実行中のタスク数と過去の完了状況を確認できます

- 各トランスコーディングタスクの実行ログとトリガーペイロードをトレースできます

- トランスコーディング関数に異常が発生した場合、デッドレター関数の実行がトリガーされます。この関数にアラームなどのカスタムロジックを追加できます

トランスコーディング完了後、OSS コンソールにログインして、指定された出力先ディレクトリでトランスコーディング済みの動画を確認できます。