E-MapReduce:リアルタイムコンピュート for Apache FlinkとPaimonを使用して、バッチ処理とストリーム処理の統合を実現する

ステップ1：リアルタイムコンピュート for Apache FlinkでPaimonカタログを作成する

Apache Paimonカタログを使用すると、同じウェアハウスディレクトリにあるすべてのApache Paimonテーブルを効率的に管理できます。 Apache Paimonカタログは、他のAlibaba Cloudサービスでも使用できます。 Apache Paimonカタログの作成と使用方法については、「Apache Paimonカタログの管理」をご参照ください。

1. リアルタイムコンピュート for Apache Flinkの管理コンソールにログオンします。

2. 目的のワークスペースを見つけ、[アクション] 列の [コンソール] をクリックします。

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

   1. 左側のナビゲーションペインで、[開発] > [スクリプト] を選択します。

   2. スクリプトタブで、アイコンをクリックしてスクリプトを作成します。

   3. スクリプトエディタにSQLコードを入力します。

      サンプルコード：

      CREATE CATALOG `paimon` WITH (
        'type' = 'paimon',
        'metastore' = 'dlf',
        'warehouse' = '<warehouse>',
        'dlf.catalog.id' = '<dlf.catalog.id>',
        'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
        'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
        'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
        'dlf.catalog.region' = '<dlf.catalog.region>',
      );

      パラメータ	説明	必須	備考
      paimon	Apache Paimonカタログの名前。	はい	カスタム名を入力します。
      type	カタログのタイプ。	はい	値をpaimonに設定します。
      metastore	メタデータストレージタイプ。	はい	値をdlfに設定します。 DLFを使用してメタデータを一元管理し、エンジン間でシームレスな統合を実現できます。
      warehouse	データウェアハウスディレクトリ。	はい	ビジネス要件に基づいてこのパラメータを設定します。
      dlf.catalog.id	DLFデータカタログのID。	はい	DLFコンソールでデータカタログのIDを確認できます
      dlf.catalog.accessKeyId	DLFへのアクセスに使用されるAccessKey ID。	はい	AccessKeyペアの取得方法については、「AccessKeyの作成」をご参照ください。
      dlf.catalog.accessKeySecret	DLFへのアクセスに使用されるAccessKeyシークレット。	はい	AccessKeyペアの取得方法については、「AccessKeyの作成」をご参照ください。
      dlf.catalog.endpoint	DLFのエンドポイント。	はい	詳細については、「サポートされているリージョンとエンドポイント」をご参照ください。 説明 DLFがリアルタイムコンピュート for Apache Flinkと同じリージョンにある場合、VPCエンドポイントが使用されます。それ以外の場合、パブリックエンドポイントが使用されます。
      dlf.catalog.region	DLFが存在するリージョン。	はい	詳細については、「サポートされているリージョンとエンドポイント」をご参照ください。 説明 このパラメータの値が、dlf.catalog.endpointパラメータで指定されたエンドポイントと一致していることを確認してください。

   4. セッションクラスタを選択または作成します。

      スクリプト編集ページの右下隅にある 環境 ドロップダウンリストから、セッションクラスタを選択します。 Ververica Runtime（VVR） 8.0.4以降を使用するセッションクラスタを選択できます。使用可能なセッションクラスタがない場合は、セッションクラスタを作成します。セッションクラスタの作成方法については、「デプロイメントのデバッグ」トピックの「ステップ1：セッションクラスタを作成する」セクションをご参照ください。

   5. 実行するコードを選択し、コードの左側にある [実行] をクリックします。

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

   [スクリプト] エディタで、次のコードを入力して選択します。次に、スクリプトエディタの左上隅にある [実行] をクリックします。

   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. 左側のナビゲーションペインで、[開発] > [ETL] を選択します。

      2. ドラフト タブで、[新規（ドラフトの作成）] アイコンをクリックします。

      3. [新規ドラフト] ダイアログボックスの SQLスクリプト タブで、[空のストリームドラフト] をクリックします。

      4. [次へ] をクリックします。

      5. [新規ドラフト] ダイアログボックスで、ドラフトのパラメータを設定します。次の表にパラメータを示します。

         パラメータ	説明
         [名前]	作成するドラフトの名前。 説明 ドラフト名は、現在のプロジェクト内で一意である必要があります。
         [場所]	ドラフトのコードファイルが保存されるフォルダ。 既存のフォルダの右側にあるアイコンをクリックして、サブフォルダを作成することもできます。
         [エンジンバージョン]	ドラフトで使用されるFlinkのエンジンバージョン。エンジンバージョン、バージョンマッピング、各バージョンのライフサイクルにおける重要な時点については、「エンジンバージョン」をご参照ください。

      6. [作成] をクリックします。

   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：EMR Serverless Spark で SQL セッションを作成する

SQL開発とクエリのためにSQLセッションを作成する必要があります。セッションの詳細については、「セッションの管理」をご参照ください。

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

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

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

   3. [spark] ページで、管理するワークスペースの名前をクリックします。

   4. [EMR Serverless Spark] ページの左側のナビゲーションペインで、[オペレーションセンター] > [セッション] を選択します。

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

   1. [SQLセッション] タブで、[SQLセッションの作成] をクリックします。

   2. [SQLセッションの作成] ページで、パラメータを設定し、[作成] をクリックします。次の表にパラメータを示します。

      パラメータ	説明
      [名前]	SQLセッションの名前。例：paimon_compute。
      [spark設定]	Paimonに接続するには、次のコードを入力します。 spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions spark.sql.catalog.paimon org.apache.paimon.spark.SparkCatalog spark.sql.catalog.paimon.metastore dlf spark.sql.catalog.paimon.warehouse <warehouse> spark.sql.catalog.paimon.dlf.catalog.id <dlf.catalog.id> ビジネス要件に基づいて、次のパラメータを設定します。 <warehouse>：データウェアハウスディレクトリ。 <dlf.catalog.id>：DLFデータカタログのID。

   3. [アクション] 列の [開始] をクリックします。

ステップ 3：EMR Serverless Spark でインタラクティブクエリを実行し、ジョブをスケジュールする

EMR Serverless Sparkでは、インタラクティブクエリを実行し、ワークフローでジョブをスケジュールして、さまざまな要件に対応できます。インタラクティブクエリを実行して、高速クエリとデバッグを実装できます。また、ワークフローを作成してジョブを開発、公開、および保守することもできます。これは、ジョブの完全なライフサイクル管理を実装するのに役立ちます。

データ書き込みプロセス中に、EMR Serverless Sparkを使用して、いつでもPaimonテーブルでインタラクティブクエリを実行できます。これにより、データの状態をリアルタイムで取得し、データの高速分析を実行できます。開発済みのジョブを公開し、ジョブに基づいてワークフローを作成してジョブを調整し、ワークフローを公開できます。スケジューリングポリシーを設定して、ジョブを定期的にスケジュールできます。これにより、データの処理と分析を自動的かつ効率的に実行できます。