AnalyticDB：使用OSS Foreign Table訪問Iceberg資料

步驟一：登入EMR執行個體

使用SSH方式登入EMR叢集。

步驟二：登入Spark SQL

執行以下命令，通過Spark SQL讀寫Iceberg配置。

spark-sql       --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
                --conf spark.sql.catalog.hadoop=org.apache.iceberg.spark.SparkCatalog \
                --conf spark.sql.catalog.hadoop.type=hadoop \
                --conf spark.sql.catalog.hadoop.warehouse=oss://testBucketName/warehouse

當出現如下資訊時，表示已進入spark-sql命令列。

spark-sql>

步驟三：建立Iceberg表並寫入資料

建立hadoop資料庫。

CREATE DATABASE IF NOT EXISTS hadoop.hadoop_db;

建立測試表並寫入資料。

-- 建立測試表
CREATE TABLE IF NOT EXISTS hadoop.hadoop_db.hadoop_sample(
    id BIGINT COMMENT 'unique id', 
    data STRING
) 
    USING iceberg;
-- 寫入資料
INSERT INTO hadoop.hadoop_db.hadoop_sample VALUES (1, 'a'), (2, 'b'), (3, 'c');

查詢資料。

SELECT * FROM hadoop.hadoop_db.hadoop_sample;

步驟四：驗證OSS檔案結構

登入OSS控制台，在目標Bucket中確認Iceberg中繼資料及資料檔案結構如下。

oss://testBucketName/warehouse/hadoop_db/hadoop_sample/  
├── metadata/
│   ├── metadata.json  
│   ├── snapshots.avro  
│   └── manifests.avro
└── data/  
    └── parquet files...

步驟五：建立Iceberg外表

根據EMR建立的Iceberg表結構，在AnalyticDB for PostgreSQL執行以下SQL建立Iceberg外表。

-- 建立OSS SERVER
CREATE SERVER oss_hadoop_srv
    FOREIGN DATA WRAPPER oss_fdw
    OPTIONS (
        endpoint 'oss-cn-hangzhou-********.aliyuncs.com', 
        bucket 'testBucketName',
        catalog_type 'hadoop',
        warehouse 'oss://testBucketName/warehouse'
  );
-- 建立OSS User Mapping
CREATE USER MAPPING FOR PUBLIC
    SERVER oss_hadoop_srv
    OPTIONS (
        id 'LTAI****************',           -- 訪問OSS的AccessKey ID
        key 'yourAccessKeySecret'       -- 訪問OSS的AccessKey Secret
  );
-- 建立OSS FDW
CREATE FOREIGN TABLE sample (
    id BIGINT,
    data text
  )
    SERVER oss_hadoop_srv 
    OPTIONS (
        format 'iceberg',
        database_name 'hadoop_db',
        table_name 'hadoop_sample'
  );

步驟六：執行聯邦查詢

使用如下SQL查詢資料。

SELECT * FROM sample;

樣本二：訪問EMR的Iceberg資料（Hive Catalog）

spark-sql       --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
                --conf spark.sql.catalog.hive=org.apache.iceberg.spark.SparkCatalog \
                --conf spark.sql.catalog.hive.type=hive

spark-sql>

CREATE DATABASE IF NOT EXISTS hive.hive_db
    COMMENT '資料庫儲存在OSS上'
    LOCATION 'oss://testBucketName/warehouse/hive_db.db/';

-- 建立測試表
CREATE TABLE IF NOT EXISTS hive.hive_db.hive_sample(
    id BIGINT COMMENT 'unique id', 
    data STRING
) 
    USING iceberg;
-- 寫入資料
INSERT INTO hive.hive_db.hive_sample VALUES (1, 'a'), (2, 'b'), (3, 'c');

SELECT * FROM hive.hive_db.hive_sample;

oss://testBucketName/warehouse/hive_db.db/hive_sample/ 
├── metadata/
│   ├── metadata.json  
│   ├── snapshots.avro  
│   └── manifests.avro
└── data/  
    └── parquet files...

-- 建立OSS SERVER
CREATE SERVER oss_hive_srv
    FOREIGN DATA WRAPPER oss_fdw
    OPTIONS (
        endpoint 'oss-cn-hangzhou-********.aliyuncs.com', 
        bucket 'testBucketName',
        catalog_type 'hive',
        hms_uris 'thrift://192.168.XXX.XXX:9083'
);
-- 建立OSS User Mapping
CREATE USER MAPPING FOR PUBLIC
    SERVER oss_hive_srv
    OPTIONS (
        id 'LTAI****************',           -- 訪問OSS的AccessKey ID
        key 'yourAccessKeySecret'       -- 訪問OSS的AccessKey Secret
    );
-- 建立OSS FDW
CREATE FOREIGN TABLE hive_sample(
    id BIGINT,
    data text
)
    SERVER oss_hive_srv OPTIONS (
        format 'iceberg',
        database_name 'hive_db',
        table_name 'hive_sample'
);

SELECT * FROM hive_sample;

步驟一：建立Iceberg外表

根據DMS OneMeta中建立的Iceberg表結構，在AnalyticDB for PostgreSQL執行以下SQL建立Iceberg外表。

-- 建立SERVER
CREATE SERVER oss_dms_serv
    FOREIGN DATA WRAPPER oss_fdw
    OPTIONS (
        endpoint 'oss-cn-hangzhou-********.aliyuncs.com',
        bucket 'testBucketName',
        catalog_type 'onemeta',
        dms_endpoint '<dms_endpoint_name>',
        dms_region '<dms_region_id>'
  );
-- 建立OSS User Mapping
CREATE USER MAPPING FOR PUBLIC
    SERVER oss_dms_serv
    OPTIONS ( 
        id 'LTAI****************',           -- 訪問OSS的AccessKey ID
        key 'yourAccessKeySecret',       -- 訪問OSS的AccessKey Secret
        dms_id 'LTAI****************',       -- 訪問DMS的AccessKey ID
        dms_key 'yourAccessKeySecret'   -- 訪問DMS的AccessKey Secret
    );
-- 建立外表
CREATE FOREIGN TABLE sample(
    id BIGINT,
    data text
)
    SERVER oss_dms_serv OPTIONS (
        format 'iceberg',
        dms_catalog_name '<dms-catalog-name>',
        database_name '<dms-database-name>',
        table_name '<dms-table-name>'
);

步驟二：執行聯邦查詢

使用如下SQL查詢資料。

SELECT * FROM sample;