对象存储 OSS：通过Flink访问OSS Tables

下载依赖JAR包

将以下 JAR 包放入 Flink 的 $FLINK_HOME/lib 目录，或在提交作业时通过 -C 参数指定。

配置环境变量

Iceberg REST Catalog 使用 SigV4 签名认证，S3FileIO 访问数据面也需要凭证。推荐通过环境变量统一传递，在启动 Flink 之前设置以下环境变量：

export AWS_ACCESS_KEY_ID=<阿里云AccessKey ID>
export AWS_SECRET_ACCESS_KEY=<阿里云AccessKey Secret>
export AWS_REGION=<地域，例如cn-hangzhou>
export AWS_DEFAULT_REGION=<地域，例如cn-hangzhou>
# 可选，使用STS临时凭证时配置
export AWS_SESSION_TOKEN=<阿里云STS TOKEN>

env.java.opts.all: "-Daws.requestChecksumCalculation=when_required -Daws.responseChecksumValidation=when_required"

方式一：使用ossutil

1. 安装或升级 ossutil

请安装ossutil 2.3.0以上版本，如已安装 ossutil，可执行以下命令升级到最新版本：

ossutil update -f

2. 配置凭证

执行 ossutil config 命令，按提示输入 AccessKey ID、AccessKey Secret 和 Region。

3. 创建 Table Bucket

ossutil tables-api create-table-bucket --name {table bucket名称} --endpoint http://{endpint} --region {region}

命令执行成功后，返回结果中包含 Table Bucket ARN，请记录该值。

4. 创建 Namespace

ossutil tables-api create-namespace --table-bucket-arn {Table Bucket ARN} --namespace {Namespace名称} --endpoint http://{endpint}

5. 创建 Table

您可以选择以下任一方式创建 Iceberg 表：

• 通过其他计算引擎创建（如 Spark）。

• 通过 ossutil 创建：先将表 schema 保存为 JSON 文件，再调用 create-table。

  以下示例的 schema 文件 schema.json 定义了 3 个字段：

  {
    "iceberg": {
      "schema": {
        "fields": [
          {"name": "event_id", "type": "string", "required": true},
          {"name": "event_time", "type": "string"},
          {"name": "event_type", "type": "string"}
        ]
      }
    }
  }

  基于 schema 文件创建 Table：

  ossutil tables-api create-table --table-bucket-arn {bucketArn} --namespace {namespace名称} --name  {表名称} --format ICEBERG --metadata file://{文件路径} --endpoint --endpoint http://{endpint}

方式二：使用AWS CLI

OSS Tables 兼容 S3 Tables API，也可以使用 AWS CLI 管理 Table Bucket。

1. 安装 AWS CLI

curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install

2. 配置凭证

执行 aws configure 命令，按提示输入 AccessKey ID、AccessKey Secret 和 Region。

3. 创建 Table Bucket

aws s3tables --endpoint http://{endpint} create-table-bucket --region {region} --name {table bucket名称}

命令执行成功后，返回结果中包含 Table Bucket ARN。

4. 创建 Namespace

aws s3tables --endpoint http://{endpoint} create-namespace --table-bucket-arn {Table Bucket ARN} --namespace {namespace名称}

5. 创建 Table

• 通过其他计算引擎（如 Spark）创建表

• 使用 AWS CLI 创建。使用 AWS CLI 时，先将完整的入参保存为 JSON 文件 create-table.json，再调用 create-table。

  {
      "tableBucketARN": "{BucketArn}",
      "namespace": "{namespace名称}",
      "name": "{表明}",
      "format": "ICEBERG",
      "metadata": {
          "iceberg": {
              "schema": {
                  "fields": [
                       {"name": "event_id", "type": "string","required": true},
                       {"name": "event_time", "type": "string"},
                       {"name": "event_type", "type": "string"}
                  ]
              }
          }
      }
  }

  aws s3tables --endpoint http://{endpoint} create-table --cli-input-json file://{文件路径}

6. 管理后台维护任务

OSS Tables 支持自动执行 Iceberg 表的后台维护（如文件清理、文件合并等），通过 AWS CLI 可以查询和配置维护任务。

查询 Table 维护任务状态：

aws s3tables get-table-maintenance-job-status \
   --table-bucket-arn="{bucketArn}" \
   --namespace="{namespace名称}" \
   --name="{表名}" 

配置 Bucket 级维护策略（文件清理）：

aws s3tables put-table-bucket-maintenance-configuration \
   --table-bucket-arn {tableArn} \
   --type icebergUnreferencedFileRemoval \
   --value '{"status":"enabled","settings":{"icebergUnreferencedFileRemoval":{"unreferencedDays":4,"nonCurrentDays":10}}}' 

配置 Table 级维护策略（小文件合并）：

aws s3tables put-table-maintenance-configuration \
   --table-bucket-arn {bucketArn} \
   --type icebergCompaction \
   --namespace {namespace名称} \
   --name {表名} \
   --value='{"status":"enabled","settings":{"icebergCompaction":{"targetFileSizeMB":256}}}'

3.1 配置Flink进程参数

在 flink-conf.yaml 中添加以下配置，用于 S3FileIO 的 Region 和凭证提供方式：

3.2 创建Catalog

在 Flink SQL 中执行以下语句创建 Catalog：

CREATE CATALOG mycatalog WITH (
  'type'                 = 'iceberg',
  'catalog-impl'         = 'org.apache.iceberg.rest.RESTCatalog',
  'io-impl'              = 'org.apache.iceberg.aws.s3.S3FileIO',
  'uri'                  = 'https://{tableBucketName}.{region}-internal.oss-tables.aliyuncs.com/iceberg',
  'warehouse'            = '<Table Bucket ARN>',
  'rest.sigv4-enabled'   = 'true',
  'rest.signing-name'    = 'osstables',
  'rest.signing-region'  = '<Region>',
  's3.endpoint'          = 'https://{region}-internal.oss.aliyuncs.com',
  's3.path-style-access' = 'false'
);

配置参数说明

4.4 建表示例

CREATE TABLE IF NOT EXISTS mycatalog.<Namespace>.<表名> (
  event_id    STRING,
  event_time  STRING,
  event_type  STRING
) WITH (
  'format-version'                     = '2',
  'write.format.default'               = 'parquet',
  'write.target-file-size-bytes'       = '33554432',        -- 32 MB（不要 128MB）
  'write.parquet.row-group-size-bytes' = '8388608'          -- 8 MB
);

资源定义

• Table Bucket ARN：acs:osstables:<Region>:<阿里云账号ID>:bucket/<bucket_name>

• Table ARN：acs:osstables:<Region>:<阿里云账号ID>:bucket/<bucket_name>/table/<table_id>

Action 定义

下表列出 OSS Tables 支持的 Action，及其是否支持跨账号授权：

Iceberg REST操作与权限映射

下表列出 Iceberg REST Catalog 各操作所需的 OSS Action：