Write to OSS Tables by using Kafka Connect

Download dependency JAR packages

Place the following JAR packages in the Kafka Connect plug-in directory, which is the path specified by plugin.path.

Method 1: Use ossutil

1. Install or update ossutil

Install ossutil version 2.3.0 or later. If you have already installed ossutil, run the following command to update it to the latest version:

ossutil update -f

2. Configure credentials

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

3. Create a Table Bucket

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

After the command runs successfully, the output includes the Table Bucket ARN. Record this value.

4. Create a namespace

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

5. Create a table

You can create an Apache Iceberg table by using one of the following methods:

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

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

  In the following example, the schema file schema.json defines 3 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 {table_bucket_arn} --namespace {namespace_name} --name {table_name} --format ICEBERG --metadata file://schema.json --endpoint http://{endpoint}

Method 2: Use the AWS CLI

OSS Tables is compatible with the S3 Tables API. 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 the 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 runs successfully, the output includes 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 by using another compute engine, such as Spark.

• To create the table by using the AWS CLI, first save the input parameters to a JSON file, such as create-table.json, and then call create-table.

  {
    "tableBucketARN": "{table_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 Apache Iceberg tables, such as file cleanup and small 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="{table_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_bucket_arn} \
   --type icebergUnreferencedFileRemoval \
   --value '{"status":"enabled","settings":{"icebergUnreferencedFileRemoval":{"unreferencedDays":4,"nonCurrentDays":10}}}' 

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

aws s3tables put-table-maintenance-configuration \
   --table-bucket-arn {table_bucket_arn} \
   --type icebergCompaction \
   --namespace {namespace_name} \
   --name {table_name} \
   --value='{"status":"enabled","settings":{"icebergCompaction":{"targetFileSizeMB":256}}}'

Connector configuration

When you create an Iceberg Sink Connector, specify the connector class as org.apache.iceberg.connect.IcebergSinkConnector and configure the following properties:

# --- Iceberg Catalog (REST) ---
iceberg.catalog.type: rest
iceberg.catalog.uri: https://{region}-internal.oss-tables.aliyuncs.com/iceberg
iceberg.catalog.rest.sigv4-enabled: true
iceberg.catalog.rest.signing-region: {region}
iceberg.catalog.warehouse: {table_bucket_arn}
iceberg.catalog.rest.signing-name: osstables
iceberg.catalog.rest.access-key-id: {access_key_id}
iceberg.catalog.rest.secret-access-key: {access_key_secret}


# --- Force S3FileIO (catalog returns oss:// but storage is S3-compatible) ---
iceberg.catalog.io-impl: org.apache.iceberg.aws.s3.S3FileIO

# --- S3FileIO storage configuration ---
iceberg.catalog.s3.endpoint: https://oss-{region}-internal.aliyuncs.com
iceberg.catalog.s3.access-key-id: {access_key_id}
iceberg.catalog.s3.secret-access-key: {access_key_secret}
iceberg.catalog.s3.path-style-access: true
iceberg.catalog.client.region: {region}

# --- Data format conversion ---
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: false
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: false

-Daws.requestChecksumCalculation=when_required
-Daws.responseChecksumValidation=when_required

Parameters

Resource definitions

• Table Bucket ARN: acs:osstables:{region}:{alibaba_cloud_account_id}:bucket/{table_bucket_name}

• Table ARN: acs:osstables:{region}:{alibaba_cloud_account_id}:bucket/{table_bucket_name}/table/{table_id}

Action definitions

The following table lists the OSS Tables actions and indicates which ones support cross-account authorization.

Permission mapping for REST operations

The following table lists the required OSS actions for each Iceberg REST Catalog operation.