DocsICP FilingConsole
官方文档
首页

Access OSS Tables from Flink

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

This topic describes how to access OSS Tables from Flink by using Iceberg REST Catalog, which is compatible with the standard Iceberg REST protocol and requires no additional deployment. You can use Flink SQL to create tables, write data, and run queries for streaming or batch data lake ingestion.

Step 2: Create a table bucket

Before writing data, you must create a table bucket and a namespace. You can use ossutil or the AWS CLI.

Method 1: Use ossutil

1. Install or upgrade ossutil

Install ossutil version 2.3.0 or later. If you have an older version installed, run the following command to upgrade it:

ossutil update -f

2. Configure credentials

Run the ossutil config command and, when prompted, enter your AccessKey ID, AccessKey Secret, and region.

3. Create a table bucket 

ossutil tables-api create-table-bucket --name {table_bucket_name} --endpoint http://{endpoint} --region {region}

After the command succeeds, the command output contains the table bucket ARN. Record this value for later use.

4. Create a namespace 

ossutil tables-api create-namespace --table-bucket-arn {table_bucket_arn} --namespace {namespace_name} --endpoint http://{endpoint}
Important

namespace and table names cannot contain hyphens (-). Use underscores (_) instead, because these names are used as identifiers in SQL statements.

5. Create a table

You can create an Iceberg table in either of the following ways:

  • Use another compute engine, such as Spark, to create the table.

  • Use ossutil to create the table. First, save the table schema to a JSON file, and then call create-table.

    The following example shows a schema file named schema.json that defines three fields:

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

    Create a table based on the schema file:

    ossutil tables-api create-table --table-bucket-arn {bucket_arn} --namespace {namespace_name} --name  {table_name} --format ICEBERG --metadata file://{file_path} --endpoint http://{endpoint}

Method 2: Use the AWS CLI

OSS Tables is compatible with the S3 Tables API, so you can also use the AWS CLI to manage a table bucket.

1. Install the AWS CLI 

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

2. Configure credentials

Run the aws configure command and enter your AccessKey ID, AccessKey Secret, and region as prompted.

3. Create a table bucket 

aws s3tables --endpoint http://{endpoint} create-table-bucket --region {region} --name {table_bucket_name}

After the command succeeds, the output contains the table bucket ARN.

4. Create a namespace 

aws s3tables --endpoint http://{endpoint} create-namespace --table-bucket-arn {table_bucket_arn} --namespace {namespace_name}

5. Create a table

  • Create the table with another compute engine, such as Spark.

  • To create the table with the AWS CLI, first save the full input parameters to a JSON file named create-table.json, and then call create-table.

    {
    "tableBucketARN": "{bucket_arn}",
    "namespace": "{namespace_name}",
    "name": "{table_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://{file_path}

6. Manage background maintenance jobs

OSS Tables supports automatic background maintenance for Iceberg tables, such as file cleanup and file compaction. You can use the AWS CLI to query and configure these maintenance jobs.

Query the status of a table maintenance job: 

aws s3tables get-table-maintenance-job-status \
   --table-bucket-arn="{bucket_arn}" \
   --namespace="{namespace_name}" \
   --name="{table_name}"

Configure a bucket-level maintenance policy (file cleanup): 

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

Configure a table-level maintenance policy (file compaction): 

aws s3tables put-table-maintenance-configuration \
   --table-bucket-arn {bucket_arn} \
   --type icebergCompaction \
   --namespace {namespace_name} \
   --name {table_name} \
   --value='{"status":"enabled","settings":{"icebergCompaction":{"targetFileSizeMB":256}}}'
Previous：Access OSS Tables by using Kafka ConnectNext：Access OSS Tables with Trino