DocsICP FilingConsole
官方文档
首页

Access OSS Tables with Spark

更新时间：
复制 MD 格式
产品详情
我的收藏

OSS Tables is compatible with the Apache Iceberg REST Catalog protocol, eliminating the need for an external catalog service like Hive Metastore. This lets compute engines like Spark connect directly to OSS Tables to create tables and manage data using standard SQL.

Prerequisites

  • You have installed Spark 3.5 or later and Java 11 or later.

  • You have created a Table Bucket in OSS Tables. For more information, see OSS Tables.

Step 1: Prepare the environment

Download dependency JARs

Place the following JAR packages in the $SPARK_HOME/jars directory, or specify them at startup using the --jars option.

JAR package

Description

iceberg-spark-runtime-3.5_2.12-1.10.1.jar

The Iceberg Spark runtime package provides Spark integration for Iceberg. Select the package that matches your Spark version. For example, use iceberg-spark-runtime-3.5_2.12 for Spark 3.5.

iceberg-aws-bundle-1.10.1.jar

The Iceberg AWS integration package provides the S3FileIO implementation and the AWS SDK required for sigv4 signature authentication with the REST Catalog. The version must match the runtime package.

Configure environment variables

Set the following environment variables before starting Spark to provide credentials for the REST Catalog and S3FileIO:

Note

The environment variables use the AWS_ prefix because Iceberg's sigv4 signing module and S3FileIO reuse the standard credential provider chain in the AWS SDK. Use your Alibaba Cloud AccessKey ID and AccessKey Secret for the values.

export AWS_ACCESS_KEY_ID=<Alibaba Cloud AccessKey ID>
export AWS_SECRET_ACCESS_KEY=<Alibaba Cloud AccessKey Secret>
export AWS_REGION=<region, for example, cn-hangzhou>
# Optional. Configure this when using STS temporary credentials.
export AWS_SESSION_TOKEN=<Alibaba Cloud STS TOKEN>
Important

If you use environment variables, ensure that both the driver and the executors can access them. In YARN mode, inject the environment variables on each cluster node. In Kubernetes mode, inject them into the pods. If this is not feasible, specify credentials explicitly using Spark configuration properties as described below.

Configure Spark configuration properties

In addition to environment variables, you can provide credentials explicitly using Spark catalog properties in PySpark or Spark SQL. This method is better for multi-catalog scenarios or when setting environment variables is impractical.

# S3FileIO data plane credentials
spark.sql.catalog.oss_tables.s3.access-key-id=<Alibaba Cloud AccessKey ID>
spark.sql.catalog.oss_tables.s3.secret-access-key=<Alibaba Cloud AccessKey Secret>
spark.sql.catalog.oss_tables.client.region=<region, for example, cn-hangzhou>
# Optional. Configure this when using STS temporary credentials.
spark.sql.catalog.oss_tables.s3.session-token=<Alibaba Cloud STS TOKEN>

# REST Catalog signing credentials
spark.sql.catalog.oss_tables.rest.access-key-id=<Alibaba Cloud AccessKey ID>
spark.sql.catalog.oss_tables.rest.secret-access-key=<Alibaba Cloud AccessKey Secret>
spark.sql.catalog.oss_tables.rest.signing-region=<region, for example, cn-hangzhou>
# Optional. Configure this when using STS temporary credentials.
spark.sql.catalog.oss_tables.rest.session-token=<Alibaba Cloud STS TOKEN>

Step 2: Configure Spark connection

  • OSS Tables provides an Iceberg REST Catalog endpoint that Spark uses to manage table metadata. The endpoint formats are as follows:

    • Internal network: https://{region}-internal.oss-tables.aliyuncs.com/iceberg

    • Public network: https://{region}.oss-tables.aliyuncs.com/iceberg

  • OSS Tables provides an S3FileIO endpoint that Spark uses to access table data. The endpoint formats are as follows:

    • Internal network: https://oss-{region}-internal.aliyuncs.com

    • Public network: https://oss-{region}.aliyuncs.com

Note

Do not set io-impl to org.apache.iceberg.hadoop.HadoopFileIO.

Iceberg is designed to avoid list operations. As a storage type deeply optimized for Iceberg, Table Bucket prohibits these operations. HadoopFileIO is implemented based on the file semantics of object storage and triggers list operations for compatibility with general use cases, making it unsuitable for data access in an OSS Table Bucket.

Start with PySpark

The following example uses a Table Bucket named my-data-lake in the cn-hangzhou region. The corresponding Table Bucket ARN is acs:osstables:cn-hangzhou:{accountId}:bucket/my-data-lake.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("OSS Tables Demo") \
    .config("spark.jars", "/path/to/iceberg-spark-runtime-3.5_2.12-1.10.1.jar,"
            "/path/to/iceberg-aws-bundle-1.10.1.jar") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.oss_tables", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.oss_tables.catalog-impl", "org.apache.iceberg.rest.RESTCatalog") \
    .config("spark.sql.catalog.oss_tables.uri", "https://cn-hangzhou-internal.oss-tables.aliyuncs.com/iceberg") \
    .config("spark.sql.catalog.oss_tables.warehouse", "acs:osstables:cn-hangzhou:{accountId}:bucket/my-data-lake") \
    .config("spark.sql.catalog.oss_tables.rest.sigv4-enabled", "true") \
    .config("spark.sql.catalog.oss_tables.rest.signing-region", "cn-hangzhou") \
    .config("spark.sql.catalog.oss_tables.rest.signing-name", "osstables") \
    .config("spark.sql.catalog.oss_tables.io-impl", "org.apache.iceberg.aws.s3.S3FileIO") \
    .config("spark.sql.catalog.oss_tables.s3.endpoint", "https://oss-cn-hangzhou-internal.aliyuncs.com") \
    .getOrCreate()

Start with spark-sql

spark-sql \
  --jars /path/to/iceberg-spark-runtime-3.5_2.12-1.10.1.jar,/path/to/iceberg-aws-bundle-1.10.1.jar \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.oss_tables=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.oss_tables.catalog-impl=org.apache.iceberg.rest.RESTCatalog \
  --conf spark.sql.catalog.oss_tables.uri=https://cn-hangzhou-internal.oss-tables.aliyuncs.com/iceberg \
  --conf spark.sql.catalog.oss_tables.warehouse=acs:osstables:cn-hangzhou:{accountId}:bucket/my-data-lake \
  --conf spark.sql.catalog.oss_tables.rest.sigv4-enabled=true \
  --conf spark.sql.catalog.oss_tables.rest.signing-region=cn-hangzhou \
  --conf spark.sql.catalog.oss_tables.rest.signing-name=osstables \
  --conf spark.sql.catalog.oss_tables.io-impl=org.apache.iceberg.aws.s3.S3FileIO \
  --conf spark.sql.catalog.oss_tables.s3.endpoint=https://oss-cn-hangzhou-internal.aliyuncs.com

Parameters

Parameter

Required

Description

catalog-impl

Yes

Set to org.apache.iceberg.rest.RESTCatalog to use the REST Catalog.

uri

Yes

The URL of the REST Catalog endpoint. Formats:

  • Internal network: https://{region}-internal.oss-tables.aliyuncs.com/iceberg

  • Public network: https://{region}.oss-tables.aliyuncs.com/iceberg

warehouse

Yes

The Alibaba Cloud Resource Name (ARN) of the Table Bucket. Format: acs:osstables:{region}:{accountId}:bucket/{bucket-name}.

rest.sigv4-enabled

Yes

Set to true to enable sigv4 signature authentication.

rest.signing-region

No (if you configure credentials by using environment variables)

The region for REST Catalog signing. It must be the same as the region where the Table Bucket is located, for example, cn-hangzhou. This parameter can be omitted if you configure credentials by using the AWS_REGION environment variable. It is required if you provide credentials through Spark properties.

rest.signing-name

Yes

The signing service name. Must be set to osstables.

io-impl

Yes

The FileIO implementation for reading and writing data files. Set to org.apache.iceberg.aws.s3.S3FileIO.

s3.endpoint

Yes

The S3FileIO endpoint for accessing the OSS data plane. It must include the https:// prefix. Formats:

  • Internal network: https://oss-{region}-internal.aliyuncs.com

  • Public network: https://oss-{region}.aliyuncs.com

Step 3: Manage data with SQL

After establishing a connection, use standard SQL to manage your data.

Manage namespaces

A namespace logically groups tables and is similar to a database.

-- View existing namespaces
SHOW NAMESPACES IN oss_tables;

-- Create a namespace
CREATE NAMESPACE oss_tables.my_namespace;

-- Drop a namespace (you must drop all tables in it first)
DROP NAMESPACE oss_tables.my_namespace;

Create and manage tables

-- Create a non-partitioned table
CREATE TABLE oss_tables.my_namespace.users (
    id BIGINT NOT NULL COMMENT 'User ID',
    name STRING COMMENT 'Username',
    email STRING COMMENT 'Email address',
    created_at TIMESTAMP COMMENT 'Creation time'
) USING iceberg;

-- Create a partitioned table (partitioned by day)
CREATE TABLE oss_tables.my_namespace.events (
    id BIGINT NOT NULL,
    event_type STRING,
    data STRING,
    ts TIMESTAMP
) USING iceberg
PARTITIONED BY (days(ts));

-- View all tables in a namespace
SHOW TABLES IN oss_tables.my_namespace;

-- View the table schema
DESCRIBE TABLE oss_tables.my_namespace.users;

-- Drop a table. OSS Tables requires the PURGE keyword.
DROP TABLE oss_tables.my_namespace.users PURGE;

Write and query data

-- Insert data
INSERT INTO oss_tables.my_namespace.users VALUES
    (1, 'Alice', 'alice@example.com', TIMESTAMP '2024-01-15 10:30:00'),
    (2, 'Bob', 'bob@example.com', TIMESTAMP '2024-01-16 14:20:00'),
    (3, 'Charlie', 'charlie@example.com', TIMESTAMP '2024-01-17 09:15:00');

-- Query all data from the table
SELECT * FROM oss_tables.my_namespace.users;

-- Query with a condition
SELECT * FROM oss_tables.my_namespace.users WHERE id = 2;

-- Aggregate query
SELECT COUNT(*) AS total FROM oss_tables.my_namespace.users;

-- Grouped aggregation
SELECT name, COUNT(*) AS cnt FROM oss_tables.my_namespace.users GROUP BY name;

-- Update data
UPDATE oss_tables.my_namespace.users SET name = 'David' WHERE id = 3;

-- Delete data
DELETE FROM oss_tables.my_namespace.users WHERE id = 1;

-- Verify the result
SELECT * FROM oss_tables.my_namespace.users ORDER BY id;

Partitioned table operations

-- Insert partitioned data
INSERT INTO oss_tables.my_namespace.events VALUES
    (1, 'click', '{"page": "home"}', TIMESTAMP '2024-01-15 10:30:00'),
    (2, 'view', '{"page": "product"}', TIMESTAMP '2024-01-15 11:00:00'),
    (3, 'click', '{"page": "detail"}', TIMESTAMP '2024-01-16 09:00:00');

-- Query with partition pruning (scans only matching partitions)
SELECT * FROM oss_tables.my_namespace.events
WHERE ts >= TIMESTAMP '2024-01-15 00:00:00'
  AND ts < TIMESTAMP '2024-01-16 00:00:00';

-- Aggregate statistics
SELECT event_type, COUNT(*) AS cnt
FROM oss_tables.my_namespace.events
GROUP BY event_type;

Time travel queries

Iceberg supports time travel, which lets you query data from a specific snapshot or point in time.

-- View the snapshot history
SELECT snapshot_id, committed_at, operation
FROM oss_tables.my_namespace.users.snapshots;

-- Query historical data by snapshot ID
SELECT * FROM oss_tables.my_namespace.users
VERSION AS OF <snapshot_id>;

-- Query data at a specific point in time
SELECT * FROM oss_tables.my_namespace.users
TIMESTAMP AS OF TIMESTAMP '2024-01-16 00:00:00';

-- View the data file distribution
SELECT * FROM oss_tables.my_namespace.users.files;

Usage notes

  • Version requirements: Use Spark 3.5 or later and Iceberg 1.10.1. The Spark version must match the version of the iceberg-spark-runtime JAR package. For example, use iceberg-spark-runtime-3.5_2.12 for Spark 3.5.

  • Credential configuration: S3FileIO and the REST Catalog sigv4 signing share the same AccessKey ID and AccessKey Secret. Configure them using the following environment variables, which apply to both services:

    • AWS_ACCESS_KEY_ID: Your Alibaba Cloud AccessKey ID

    • AWS_SECRET_ACCESS_KEY: Your Alibaba Cloud AccessKey Secret

    • AWS_REGION: The region where your Table Bucket is located. This is used for REST Catalog signing. If you set this environment variable, you can omit the rest.signing-region property in your catalog configuration.

  • Table format: OSS Tables supports only the Iceberg format. You must specify USING iceberg when creating a table.

  • Data maintenance: OSS Tables provides built-in file compaction, snapshot cleanup, and unreferenced file cleanup. You do not need to run Iceberg maintenance procedures in Spark. For more information, see data maintenance.

  • Supported OSS Tables endpoints:

    • Internal network: https://{region}-internal.oss-tables.aliyuncs.com/iceberg

    • Public network: https://{region}.oss-tables.aliyuncs.com/iceberg

Previous：Iceberg REST CatalogNext：Access OSS Tables by using Kafka Connect