使用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;