Data Lake Formation:Access DLF from EMR on ECS Spark

Step 1: Attach the RAM policy to AliyunECSInstanceForEMRRole

1. Log on to the Resource Access Management (RAM) console using your Alibaba Cloud account or as a RAM administrator.

2. In the navigation pane, choose Identity Management > Roles, then search for AliyunECSInstanceForEMRRole.

3. In the Actions column, click Add Permissions.

4. Under Permission Policies, search for and select AliyunDLFFullAccess, then click Confirm.

   

Step 2: Grant DLF permissions to AliyunECSInstanceForEMRRole

1. Log on to the Data Lake Formation console.

2. On the Catalogs page, click the name of the catalog to open its details page.

3. Click the Permissions tab to grant permissions at the catalog level. To grant permissions at a lower scope, navigate to the target database or table and click its Permissions tab instead.

4. On the authorization page, configure the following settings and click OK.

   Note

   If AliyunECSInstanceForEMRRole does not appear in the dropdown list, go to the user management page and click Sync.

   Field	Value
   User/Role	RAM User/RAM Role
   Select Authorization Object	AliyunECSInstanceForEMRRole
   Preset Permission Type	Select read permissions manually, or choose a predefined role: Data Reader or Data Editor

Step 1: Upload JAR files and the upgrade script to OSS

1. Upload both JAR files to Object Storage Service (OSS) and set their permissions to public-read. For upload instructions, see Simple upload.

2. Modify the following script by replacing the two placeholder URLs with the actual OSS download URLs of your JAR files, then upload the script to OSS.

   Important

   EMR on ECS clusters cannot access the public network by default. Use private network URLs when your OSS bucket and EMR cluster are in the same region.

   Placeholder	JAR file	URL format
   <paimon-jindo-1.1.0.jar-url>	paimon-jindo-*.jar	Private network: https://<bucket>.oss-cn-hangzhou-internal.aliyuncs.com/jars/paimon-jindo-1.1.0.jar<br>Public network: https://<bucket>.oss-cn-hangzhou.aliyuncs.com/jars/paimon-jindo-1.1.0.jar
   <paimon-spark-3.x-1.1.0.jar-url>	paimon-spark-3.x-*.jar	Private network: https://<bucket>.oss-cn-hangzhou-internal.aliyuncs.com/jars/paimon-spark-3.x-1.1.0.jar<br>Public network: https://<bucket>.oss-cn-hangzhou.aliyuncs.com/jars/paimon-spark-3.x-1.1.0.jar

   #!/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-url>
   wget <paimon-spark-3.x-1.1.0.jar-url>
   
   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

   Replace each placeholder with the OSS download URL of the corresponding JAR file:

Step 2: Run the script on your EMR cluster

Run the script as a Script Action across all nodes. For details, see Manually run a script.

1. In your EMR cluster, go to Script Action > Manual Execution and click Create And Execute.

2. In the dialog box, configure the following settings and click OK.

   Field	Value
   Name	A custom name for the script
   Script Location	The OSS path of the upgrade script. Format: oss://**/*.sh
   Execution Scope	Cluster

3. After the script completes, restart the Spark service for the changes to take effect.

Connect to the Paimon catalog

Run the following spark-sql command. Replace <regionID> with your region ID (for example, cn-hangzhou), and replace <catalog> with the name of the DLF catalog you created.

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

Key parameters:

Create tables

Run the following SQL to create a managed table and a foreign table.

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'
);

Table behavior:

Insert data

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);

Query data

SELECT * FROM user_samples;
SELECT * FROM user_samples_di;

Merge data

The following statement merges rows from user_samples_di into user_samples, matching on user_id. Matched rows are updated; unmatched rows are inserted.

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);