E-MapReduce:Flink、EMR Serverless Spark、Paimon を用いたストリームバッチ統一データレイクの構築

Realtime Compute for Apache Flink

Alibaba Cloud Realtime Compute for Apache Flink は、Apache Flink のためのフルマネージドのサーバーレスクラウドサービスです。開発、運用、保守のためのすぐに使えるワンストッププラットフォームを提供し、柔軟な課金体系を提供します。このサービスには、ジョブ開発、データデバッグ、ジョブの実行と監視、自動チューニング、インテリジェント診断などの完全なライフサイクル機能が含まれています。詳細については、「Realtime Compute for Apache Flink とは」をご参照ください。

Apache Paimon

Apache Paimon は、Flink と Spark を使用してストリームバッチ統一のためのリアルタイムレイクハウスアーキテクチャを実現する、統一データレイクストレージフォーマットです。Paimon は、レイクハウスフォーマットと Log-structured Merge-tree (LSM) テクノロジーを革新的に組み合わせ、データレイクにリアルタイムのストリーミング更新と完全なストリーム処理機能を提供します。詳細については、「Apache Paimon」をご参照ください。

ステップ 1：Paimon カタログの作成

Paimon カタログを使用すると、同じウェアハウスディレクトリ内のすべての Paimon テーブルを管理し、他の Alibaba Cloud 製品に接続できます。Paimon カタログの作成方法と使用方法の詳細については、「Paimon カタログの管理」をご参照ください。

1. Realtime Compute for Apache Flink コンソールにログインします。

2. 対象のワークスペースの操作列で、コンソールをクリックします。

3. Paimon カタログを作成します。

   UI を使用して DLF タイプの Paimon カタログを追加します。AccessKey 情報を手動で入力する必要はありません。詳細な手順については、「Paimon カタログの管理」をご参照ください。

4. Paimon テーブルを作成します。

   Script エディターで次のコマンドを入力し、コードを選択して 実行 をクリックします。

   CREATE TABLE IF NOT EXISTS `paimon`.`test_paimon_db`.`test_append_tbl`
   (
       id       STRING,
       data     STRING,
       category INT,
       ts       STRING,
       dt       STRING,
       hh       STRING
   ) PARTITIONED BY (dt, hh)
   WITH (
       'write-only' = 'true'
   );

5. ストリームジョブを作成します。

   1. 新しいジョブを作成します。

      1. 左側のナビゲーションペインで、Development > [ETL] を選択します。

      2. 新しいストリームジョブを作成します。[新規空白ストリームドラフト] ダイアログボックスで、ジョブパラメーターを設定します。

         パラメーター	説明
         [Name]	ジョブの名前。 説明 ジョブ名は現在のプロジェクト内で一意である必要があります。
         [Engine Version]	現在のジョブの Flink エンジンバージョン。エンジンバージョン番号、バージョンの互換性、ライフサイクルマイルストーンの詳細については、「エンジンバージョン」をご参照ください。

      3. create をクリックします。

   2. コードを記述します。

      新しいストリームドラフトで、datagen コネクターを使用してデータを継続的に生成し、Paimon テーブルに書き込むために、次のコードを入力します。

      CREATE TEMPORARY TABLE datagen
      (
          id        string,
          data      string,
          category  int
      )
      WITH (
          'connector' = 'datagen',
          'rows-per-second' = '100',
          'fields.category.kind' = 'random',
          'fields.category.min' = '1',
          'fields.category.max' = '10'
      );
      INSERT INTO `paimon`.`test_paimon_db`.`test_append_tbl`
      SELECT
          id,
          data,
          category,
          cast(LOCALTIMESTAMP as string) as ts,
          cast(CURRENT_DATE as string) as dt,
          cast(hour(LOCALTIMESTAMP) as string) as hh
      FROM datagen;

   3. ジョブを本番環境に公開するには、[デプロイ] をクリックします。

   4. [デプロイメント] ページで、ジョブを開始します。詳細については、「ジョブを開始する」をご参照ください。

ステップ 2：SQL セッションの作成

SQL セッションは、SQL の開発とクエリをサポートします。セッションの詳細については、「セッションマネージャー」をご参照ください。

1. [セッション] ページに移動します。

   1. EMR コンソールにログインします。

   2. 左側のナビゲーションウィンドウで、[EMR Serverless] > [Spark] を選択します。

   3. [Spark] ページで、対象のワークスペースの名前をクリックします。

   4. EMR Serverless Spark ページで、左側のナビゲーションウィンドウの Sessions をクリックします。

2. Serverless Spark ワークスペースのデータカタログに、ステップ 1 で作成した DLF カタログを追加します。

3. SQL セッションを作成します。

   1. SQL Session タブで、Connect to SQL Session をクリックします。

   2. 「SQL セッションの作成」ページで、次のパラメーターを設定し、他のパラメーターはデフォルト値のままにして、create をクリックします。

      パラメーター	説明
      [Name]	SQL セッションの任意の名前。例：paimon_compute。

   3. 「操作」列で、START をクリックします。

ステップ 3：インタラクティブクエリまたはタスクオーケストレーションの実行

EMR Serverless Spark は、さまざまなニーズに合わせて、インタラクティブクエリとタスクオーケストレーションの 2 つの操作モードを提供します。インタラクティブクエリモードは迅速なクエリとデバッグに最適で、タスクオーケストレーションモードはジョブの開発、発行、および保守の完全なライフサイクルをサポートします。

データが書き込まれている間、EMR Serverless Spark を使用して Paimon テーブルに対していつでもインタラクティブクエリを実行し、リアルタイムのデータ状態を確認したり、迅速な分析を実行したりできます。さらに、開発したジョブを発行し、ワークフローを作成してさまざまなタスクをオーケストレーションできます。スケジューリングポリシーを設定してタスクを自動化し、効率的で自動化されたデータ処理と分析を実現できます。