使用EMR Serverless Spark讀寫Paimon表-開源巨量資料平台 E-MapReduce-阿里雲

Apache Paimon是一種流批統一的湖儲存格式，支援高吞吐的寫入和低延後查詢。本文為您介紹如何在EMR Serverless Spark中實現Paimon表的讀取與寫入操作。

前提條件

已建立工作空間，詳情請參見建立工作空間。

操作流程

步驟一：建立SQL會話

進入會話管理頁面。
1. 登入E-MapReduce控制台。
2. 在左側導覽列，選擇EMR Serverless > Spark。
3. 在Spark頁面，單擊目標工作空間名稱。
4. 在EMR Serverless Spark頁面，單擊左側導覽列中的會話管理。
在SQL會話頁面，單擊建立SQL會話。

在创建 SQL 会话頁面的Spark 配置地區，配置以下資訊，單擊创建。詳情請參見管理SQL會話。

Spark對Paimon的讀寫基於Catalog，您可以根據具體情境進行選擇。更多Catalog資訊，請參見管理資料目錄。

使用資料目錄Catalog

若採用資料目錄Catalog方式，則無需在會話中配置參數，只需在数据目录頁面單擊添加数据目录，然後在SparkSQL開發中直接選擇資料目錄即可。

說明

推薦使用esr-4.3.0及以上版本、esr-3.3.0及以上版本、esr-2.7.0及以上版本的引擎。

使用自訂Catalog

DLF（原DLF 2.5）

spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

涉及參數如下所示。

參數	說明	樣本值
`spark.sql.catalog.<catalogName>`	catalog實現。	固定值：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	指定中繼資料存放區方式，設定為`rest`表示使用DLF REST API。	固定值：`rest`
`spark.sql.catalog.<catalogName>.uri`	指定DLF的URI，格式為`http://<endpoint>-vpc.dlf.aliyuncs.com`。	`http://cn-hangzhou-vpc.dlf.aliyuncs.com`
`spark.sql.catalog.<catalogName>.warehouse`	指定資料存放區路徑（Warehouse路徑）。對於 DLF，需指定為Catalog名稱。	`<catalog_name>`
`spark.sql.catalog.<catalogName>.token.provider`	指定認證提供方，DLF 使用 `dlf`。	固定值：`dlf`
`spark.sql.catalog.<catalogName>.dlf.access-key-id`	阿里雲帳號或者RAM使用者的AccessKey ID。	`<access_key_id>`
`spark.sql.catalog.<catalogName>.dlf.access-key-secret`	阿里雲帳號或者RAM使用者的AccessKey Secret。	`<access_key_secret>`

DLF-Legacy（原DLF1.0）

中繼資料儲存在DLF-Legacy（原DLF1.0）中。

spark.sql.catalog.<catalogName>                          org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id           <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint     dlf-vpc.cn-hangzhou.aliyuncs.com

涉及參數說明如下所示。

參數	說明	樣本值
`spark.sql.catalog.<catalogname>`	catalog實現。	固定值：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogname>.metastore`	指定中繼資料存放區方式，設定為`dlf`表示使用阿里雲DLF作為中繼資料存放區。	固定值：`dlf`
`spark.sql.catalog.<catalogName>.dlf.catalog.id`	指定DLF中的Catalog名稱。	`<catalog_name>`
`spark.sql.catalog.<catalogName>.dlf.catalog.endpoint`	指定DLF的訪問端點地址，需根據您所在的地區選取項目正確的 DLF 端點。	`dlf-vpc.cn-hangzhou.aliyuncs.com`

Hive MetaStore

中繼資料儲存在指定的Hive MetaStore中。

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       hive
spark.sql.catalog.<catalogName>.uri             thrift://<yourHMSUri>:<port>

涉及參數如下表所示。

參數	說明	樣本值
`spark.sql.catalog.<catalogName>`	catalog實現。	固定值：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	指定中繼資料存放區方式，設定為 `hive` 表示使用 Hive MetaStore 作為中繼資料存放區。	固定值：`hive`
`spark.sql.catalog.<catalogName>.uri`	Hive MetaStore的URI。格式為`thrift://<Hive metastore的IP地址>:9083`。 `<Hive metastore的IP地址>`為HMS服務的內網IP地址。如果您需要指定外部Metastore服務，請參見串連外部Hive Metastore Service。	`thrift://192.168..:9083`

FileSystem

中繼資料儲存在檔案系統中。

spark.sql.catalog.<catalogName>                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore       filesystem
spark.sql.catalog.<catalogName>.warehouse       oss://<yourBucketName>/warehouse

涉及參數如下表所示。

參數	說明	樣本值
`spark.sql.catalog.<catalogName>`	catalog實現。	固定值：`org.apache.paimon.spark.SparkCatalog`
`spark.sql.catalog.<catalogName>.metastore`	指定中繼資料存放區方式，設定為`filesystem`表示使用檔案系統作為中繼資料存放區。	固定值：`filesystem`
`spark.sql.catalog.<catalogName>.warehouse`	指定中繼資料存放區路徑（Warehouse 路徑）。代碼中的`<yourBucketName>`表示OSS上的Bucket名稱。	`oss://my-bucket/warehouse`

您還可以同時配置多個Catalog，例如DLF、DLF-Legacy和Hive，具體資訊請參見以下樣本。

# 配置dlf Catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       rest
spark.sql.catalog.<catalogName>.uri                             http://cn-hangzhou-vpc.dlf.aliyuncs.com
spark.sql.catalog.<catalogName>.warehouse                       <catalog_name>
spark.sql.catalog.<catalogName>.token.provider                  dlf
spark.sql.catalog.<catalogName>.dlf.access-key-id               <access_key_id>
spark.sql.catalog.<catalogName>.dlf.access-key-secret           <access_key_secret>

# 配置dlf-legacy Catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       dlf
spark.sql.catalog.<catalogName>.dlf.catalog.id                  <catalog_name>
spark.sql.catalog.<catalogName>.dlf.catalog.endpoint            dlf-vpc.cn-hangzhou.aliyuncs.com


# 配置hive1 Catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-1>:<port>

# 配置hive2 Catalog
spark.sql.catalog.<catalogName>                                 org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.<catalogName>.metastore                       hive
spark.sql.catalog.<catalogName>.uri                             thrift://<yourHMSUri-2>:<port>

步驟二：基於Paimon Catalog的表讀寫操作

進入SQL開發頁面。
在EMR Serverless Spark頁面，單擊左側導覽列中的数据开发。
在開發目錄頁簽下，單擊表徵圖。
在建立對話方塊中，輸入名稱（例如users_task），類型使用預設的SparkSQL，然後單擊確定。

複製如下代碼到新增的Spark SQL頁簽（users_task）中。

使用Paimon Catalog

-- 建立資料庫
CREATE DATABASE IF NOT EXISTS paimon.ss_paimon_db;             

-- 建立Paimon表
CREATE TABLE paimon.ss_paimon_db.paimon_tbl (id INT, name STRING) USING paimon;

-- 寫入Paimon表
INSERT INTO paimon.ss_paimon_db.paimon_tbl VALUES (1, "a"), (2, "b"), (3, "c");

-- 查詢 Paimon 表的寫入結果
SELECT * FROM paimon.ss_paimon_db.paimon_tbl ORDER BY id;

-- 刪除資料庫
DROP DATABASE paimon.ss_paimon_db CASCADE;

在資料庫下拉式清單中選擇一個資料庫，在會話下拉式清單中選擇剛剛建立的SQL會話。
單擊运行，執行任務。返回資訊如下所示。

常見問題

對錶執行DELETE、UPDATE 或 MERGE等操作時報錯，應該如何處理？

問題現象：執行 DELETE、UPDATE 或 MERGE 操作時，出現以下類似錯誤資訊。

Caused by: org.apache.spark.sql.AnalysisException: Table does not support deletes/updates/merge: <tableName>.
    at org.apache.spark.sql.errors.QueryCompilationErrors$.tableDoesNotSupportError(QueryCompilationErrors.scala:1391)

問題原因：該表的儲存格式不支援行層級更新操作，或缺少必要的Spark配置。
解決方案：
1. 檢查表類型。
  執行以下命令，確認該表是否為Paimon表。
```
SHOW CREATE TABLE <tableName>;
```
  如果輸出結果中包含 USING PAIMON，則表為 Paimon 表。如果輸出中顯示其他儲存格式（如 USING hive），則需要確認該格式是否支援行層級更新操作。
2. 確認Spark配置。
  如果所涉及的表為Paimon表，則應檢查Spark 配置資訊，以確保已添加以下配置，從而啟用對Paimon的支援。
```
spark.sql.extensions org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
```
  如果沒有添加此配置，請在Spark 配置中添加。

E-MapReduce：使用Paimon