Data Lake Formation:EMR on ECS Spark から DLF へのアクセス

前提条件

• バージョン 5.12.0 以降の EMR クラスターを作成し、コンポーネントとして Spark 3 と Paimon を選択します。他のバージョンの要件については、DingTalk グループ (106575000021) に参加して DLF 開発者にご連絡ください。

• これでDLF のクイックスタートは完了です。

• EMR と DLF が同じリージョンにあることを確認し、ご利用の EMR クラスターの VPC を DLF ホワイトリストに追加します。

カタログの作成

詳細については、「DLF の設定」をご参照ください。

ロールへの DLF 権限の付与

1. AliyunECSInstanceForEMRRole ロールに RAM 権限を付与します。このステップは、EMR プロダクトが統合された後は不要です。

   1. ご利用の Alibaba Cloud アカウントまたは RAM 管理者として Resource Access Management (RAM) コンソールにログインします。

   2. 左側のナビゲーションウィンドウで、[ID 管理] > [ロール] を選択し、AliyunECSInstanceForEMRRole ロールを検索します。

   3. [操作] 列で [権限の追加] をクリックして、権限の追加ページに移動します。

   4. [権限ポリシー] で AliyunDLFFullAccess を検索して選択し、[確認] をクリックします。

   

2. AliyunECSInstanceForEMRRole ロールに DLF 権限を付与します。

   1. Data Lake Formation コンソールにログインします。

   2. [カタログ] ページで、カタログ名をクリックして詳細ページに移動します。

   3. カタログ全体に権限を付与するには、[権限] タブをクリックします。または、特定のデータベースまたはテーブルに移動し、その [権限] タブをクリックしてアクセス権を付与することもできます。

   4. 権限付与ページで、次の設定を行い、[OK] をクリックします。

      • ユーザー/ロール: RAM ユーザー/RAM ロールを選択します。

      • [承認オブジェクトの選択]：ドロップダウンリストから AliyunECSInstanceForEMRRole を選択します。

        説明

        ドロップダウンリストに AliyunECSInstanceForEMRRole が表示されない場合は、ユーザー管理ページに移動して [同期] をクリックしてください。

      • [プリセット権限タイプ]：読み取り権限を手動で選択するか、データ閲覧者やデータ編集者などの定義済みロールを使用します。

EMR クラスター内の Paimon 依存関係のアップグレード

Maven リポジトリから Paimon バージョン 1.1 以降の JAR ファイル (paimon-jindo-*.jar と paimon-spark-3.x-*.jar) を 2 つダウンロードします。選択するバージョンが、ご利用の EMR クラスターの Spark バージョンと一致していることを確認してください。

1. Paimon 依存関係のインポート

   1. paimon-jindo-*.jar と paimon-spark-3.x-*.jar の 2 つの JAR ファイルを OSS にアップロードし、ファイルの権限を公開読み取りに設定します。詳細については、「簡易アップロード」をご参照ください。

   2. 次のスクリプトを修正して OSS にアップロードします。

      #!/bin/bash
      
      echo 'clean up paimon-dlf-2.5 exists file'
      rm -rf /opt/apps/PAIMON/paimon-dlf-2.5
      rm -rf /opt/apps/PAIMON/paimon-dlf-2.5.tar.gz.*
      cd /opt/apps/PAIMON/paimon-current/lib/spark3
      mkdir -p /opt/apps/PAIMON/paimon-dlf-2.5/lib/spark3
      cd /opt/apps/PAIMON/paimon-dlf-2.5/lib/spark3
      wget ${paimon-jindo-1.1.0.jar}
      wget ${paimon-spark-3.x-1.1.0.jar}
      
      echo 'link paimon-current to paimon-dlf-2.5'
      rm -f /opt/apps/PAIMON/paimon-current
      ln -sf /opt/apps/PAIMON/paimon-dlf-2.5 /opt/apps/PAIMON/paimon-current

      重要

      プレースホルダーの ${paimon-jindo-1.1.0.jar} と ${paimon-spark-3.x-1.1.0.jar} を実際の OSS ダウンロード URL に置き換えてください。デフォルトでは、EMR on ECS クラスターはパブリックネットワークにアクセスできません。

      • プライベートネットワーク：https://{bucket}.oss-cn-hangzhou-internal.aliyuncs.com/jars/paimon-jindo-1.1.0.jar

      • パブリックネットワーク：https://{bucket}.oss-cn-hangzhou.aliyuncs.com/jars/paimon-jindo-1.1.0.jar

2. ご利用の EMR クラスターでスクリプトをブートストラップアクションとして実行します。詳細については、「スクリプトの手動実行」をご参照ください。

   1. EMR クラスターで、[スクリプトアクション] > [手動実行] タブで、[作成して実行] をクリックします。

   2. 表示されるダイアログボックスで、次の設定を行い、[OK] をクリックします。

      • [名前]：カスタムのスクリプト名を入力します。

      • [スクリプトの場所]：OSS にアップロードしたアップグレードスクリプトを選択します。パスは oss://**/*.sh のフォーマットである必要があります。

      • 実行スコープ: [クラスター] を選択します。

3. スクリプトの実行後、Spark サービスを再起動して変更を有効にします。

Spark を使用したデータの読み書き

spark-sql --master yarn \
  --conf spark.driver.memory=5g \
  --conf spark.sql.defaultCatalog=paimon \
  --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \
  --conf spark.sql.catalog.paimon.metastore=rest \
  --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \
  --conf spark.sql.catalog.paimon.uri=http://${regionID}-vpc.dlf.aliyuncs.com \
  --conf spark.sql.catalog.paimon.warehouse=${catalog} \
  --conf spark.sql.catalog.paimon.token.provider=dlf \
  --conf spark.sql.catalog.paimon.dlf.token-loader=ecs

CREATE TABLE user_samples
(
    user_id INT,             
    age INT,           
    gender_code STRING,    
    clk BOOLEAN
);

CREATE TABLE user_samples_di (
    user_id INT,             
    age INT,           
    gender_code STRING,    
    clk BOOLEAN
)
USING CSV
OPTIONS(
'path'='oss://${bucket}/user/user_samples_di'
);

INSERT INTO user_samples VALUES
(1, 25, 'M', true),
(2, 18, 'F', false);

INSERT INTO user_samples_di VALUES
(1, 25, 'M', true),
(2, 18, 'F', true),
(3, 35, 'M', true);

SELECT * FROM user_samples;
SELECT * FROM user_samples_di;

MERGE INTO user_samples
USING user_samples_di
ON user_samples.user_id = user_samples_di.user_id
WHEN MATCHED THEN
UPDATE SET
  age = user_samples_di.age,
  gender_code = user_samples_di.gender_code,
  clk = user_samples_di.clk
WHEN NOT MATCHED THEN
  INSERT (user_id, age, gender_code, clk)
  VALUES (user_samples_di.user_id, user_samples_di.age, user_samples_di.gender_code, user_samples_di.clk);