AI 助理
文档备案控制台
开发者社区 云存储 技术博文 文章 正文

流数据还在写文件慢慢导？Kafka Connect 一条链路直接入湖【详解 OSS Tables 系列】

2026-08-19 19
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 本文详解如何通过Kafka Connect Iceberg Sink Connector，将Kafka流数据实时写入阿里云OSS Tables数据湖。涵盖环境准备、Table Bucket/命名空间/表创建、Connector配置及权限设置，实现免运维、Exactly-Once的流式入湖。

流数据的归宿问题

在 OSS Tables 的生态里，Kafka Connect 是入湖管道，把 Kafka 消息原样写入 OSS Tables，不加不改，只负责搬。Kafka 是数据管道的咽喉要道，但咽喉不是终点，事件流进去之后最终要落到数据湖。Kafka Connect 就是这条最短的路。

常见的做法是写入一个下游存储。写数据库、写搜索索引、写对象存储里的文件，各有各的理由。但当数据量大到一定程度，或者你需要把多路流汇聚起来做离线分析时，“数据湖”几乎是绕不开的终点。

问题出在“写入”这一步。传统方案要么自己写 Consumer 把消息拉出来转成 Parquet 文件塞进 OSS，要么搭一套专门的入湖管道，但不管哪种，都要处理 Schema 管理、文件滚动、Exactly-Once 语义、分区策略这些工程细节。

Kafka Connect 的 Iceberg Sink Connector 把这些问题封装了。而 OSS Tables 兼容 Iceberg REST Catalog 协议，意味着可以通过 Kafka Connect 的 Iceberg Sink Connector 将 Kafka 消息实时写入 OSS Tables 中的表，实现流式数据入湖。

步骤一：环境准备

下载依赖 JAR 包

将以下 JAR 包放入 Kafka Connect 的插件目录（plugin.path 指定的路径）。

JAR包

版本要求

说明

iceberg-aws-bundle-1.10.1.jar

匹配iceberg版本

提供 S3FileIO 实现及 REST Catalog SigV4 签名认证所需的 AWS SDK。

iceberg-aws-1.10.1.jar

匹配iceberg版本

提供 SigV4 签名和 S3FileIO 实现。版本需与 iceberg-aws-bundle 一致。

iceberg-parquet-1.10.1.jar

匹配iceberg版本

Parquet 文件格式写入支持。

hadoop-client-runtime-3.3.6.jar

3.3.6

Hadoop 运行时依赖（Iceberg 内部加载需要），版本可按需调整。

hadoop-client-api-3.3.6.jar

3.3.6

Hadoop API 依赖（Iceberg 内部加载需要），版本可按需调整。

failsafe-3.3.2.jar

3.3.2

Iceberg SnapshotProducer 运行时需要此依赖，缺少会导致 ClassNotFoundException。

步骤二：创建Table Bucket

在开始写入数据之前，需要创建 Table Bucket 和 Namespace。可以使用 ossutil 创建，AWS CLI 也支持。

方式一：使用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}

重要：Namespace 和 Table 名称不能包含连字符（-），可使用下划线（_），这是因为名称会用于 SQL 语句中的标识符。

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 <Table Bucket ARN> --namespace <Namespace名称> --name <表名称> --format ICEBERG --metadata file://schema.json --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": "<Table Bucket ARN>",
  "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}}}'

步骤三：配置 Kafka Connect

OSS Tables 提供 Iceberg REST Catalog 端点，Kafka Connect 通过 Iceberg Sink Connector 连接该端点写入数据。Endpoint格式如下：

OSS Tables 提供S3FileIO访问OSS数据面使用的访问端点，Spark 通过该端点访问表数据。Endpoint格式如下：

Connector配置

创建 Iceberg Sink Connector 时，指定 Connector 类为 org.apache.iceberg.connect.IcebergSinkConnector，并配置以下属性： 

# --- 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: <AccessKey ID>
iceberg.catalog.rest.secret-access-key: <AccessKey Secret>
# --- Force S3FileIO (catalog returns oss:// but storage is S3-compatible) ---
iceberg.catalog.io-impl: org.apache.iceberg.aws.s3.S3FileIO
# --- S3FileIO 存储配置 ---
iceberg.catalog.s3.endpoint: https://oss-{region}-internal.aliyuncs.com
iceberg.catalog.s3.access-key-id: <AccessKey ID>
iceberg.catalog.s3.secret-access-key: <AccessKey Secret>
iceberg.catalog.s3.path-style-access: true
iceberg.catalog.client.region: <Region>
# --- 数据格式转换 ---
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

重要：如果使用较高版本的 AWS SDK（2.20+），可能出现签名错误：aws-chunked encoding is not supported with the specified x-amz-content-sha256 value。此时需要在 Kafka Connect 的 Java 启动参数中添加以下 JVM 选项：

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

配置参数说明

参数

是否必填

说明

iceberg.catalog.type

是

固定为 rest，指定使用 REST Catalog。

iceberg.catalog.uri

是

REST Catalog 端点 URL。格式：

iceberg.catalog.warehouse

是

Table Bucket ARN。格式：acs:osstables:<Region>:<阿里云账号ID>:bucket/<Table Bucket名称>。

iceberg.catalog.rest.sigv4-enabled

是

固定为 true，启用 SigV4 签名认证。

iceberg.catalog.rest.signing-name

是

固定为 osstables，OSS Tables 服务端点的 SigV4 签名服务名。

iceberg.catalog.io-impl

是

固定为 org.apache.iceberg.aws.s3.S3FileIO，使用 S3 协议访问 OSS 数据面。

iceberg.catalog.s3.endpoint

是

OSS 数据面端点。格式：

iceberg.catalog.s3.path-style-access

是

固定为 true，使用 Path-Style 访问模式。

权限配置

使用 RAM 用户或 STS 临时凭证访问 OSS Tables 时，需确保对应身份具备所需的操作权限。

资源定义

  • 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，及其是否支持跨账号授权：

分类

Action

跨账号访问

Table Bucket 级别

oss:CreateTableBucket

不允许

oss:GetTableBucket

允许

oss:ListTableBuckets

不允许

oss:CreateNamespace

允许

oss:GetNamespace

允许

oss:ListNamespaces

允许

oss:DeleteNamespace

允许

oss:DeleteTableBucket

允许

oss:PutTableBucketPolicy

不允许

oss:GetTableBucketPolicy

不允许

oss:DeleteTableBucketPolicy

不允许

oss:GetTableBucketMaintenanceConfiguration

允许

oss:PutTableBucketMaintenanceConfiguration

允许

oss:PutTableBucketEncryption

不允许

oss:GetTableBucketEncryption

不允许

oss:DeleteTableBucketEncryption

不允许

Table 级别

oss:GetTableMaintenanceConfiguration

允许

oss:PutTableMaintenanceConfiguration

允许

oss:PutTablePolicy

不允许

oss:GetTablePolicy

不允许

oss:DeleteTablePolicy

不允许

oss:CreateTable

允许

oss:GetTable

允许

oss:GetTableMetadataLocation

允许

oss:ListTables

允许

oss:RenameTable

允许

oss:UpdateTableMetadataLocation

允许

oss:GetTableData

允许

oss:PutTableData

允许

oss:GetTableEncryption

不允许

oss:PutTableEncryption

不允许

oss:DeleteTable

允许

Iceberg REST操作与权限映射

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

Iceberg REST 操作

所需 OSS Action

getConfig

oss:GetTableBucket

listNamespaces

oss:ListNamespaces

createNamespace

oss:CreateNamespace

loadNamespaceMetadata

oss:GetNamespace

dropNamespace

oss:DeleteNamespace

listTables

oss:ListTables

createTable

oss:CreateTable、oss:PutTableData

loadTable

oss:GetTableMetadataLocation、oss:GetTableData

updateTable

oss:UpdateTableMetadataLocation、oss:PutTableData、oss:GetTableData

dropTable

oss:DeleteTable

renameTable

oss:RenameTable

tableExists

oss:GetTable

namespaceExists

oss:GetNamespace
文章标签：
对象存储
消息中间件
Kafka
网络架构
Java
阿里云存储
目录
相关文章
人工智能 缓存 前端开发
6540 22
人工智能 JavaScript 开发工具
3390 6
缓存 JavaScript Shell
1604 2
开发工具 Swift git
1273 1
Shell API 调度
902 2
云计算学习者
|
14天前
|
存储 弹性计算 缓存
阿里云服务器租赁费用：新版租赁收费标准及活动报价参考
本文更新了2026年阿里云全系列云服务器租赁活动报价，所有特惠资源均可前往阿里云活动中心选购，整体覆盖从个人入门到企业级高性能场景的全梯度需求。其中轻量应用服务器主打极致性价比，2核2G峰值200M带宽配置每日10点、15点限时抢购价仅38元/年，2核4G配置379元/年起；高性价比的经济型e实例、通用算力型u2i实例覆盖2核4G至4核32G全档位，适配开发测试与中小型企业业务；搭载英特尔至强6处理器的第九代c9i企业级实例算力较上代提升20%，支撑高并发生产环境，不同实例规格价差清晰，用户可根据自身业务负载与预算灵活选型。
云计算学习者
2151 121
阿里云服务器租赁费用：新版租赁收费标准及活动报价参考
程序员天天困
|
15天前
|
人工智能 程序员 API
Codex 接入 DeepSeek-V4-Flash：还能补上识图，提供两套方案
Codex 接入 DeepSeek-V4-Flash 怎么配？本文覆盖 CLI 与桌面端，再用 qwen3-vl-flash 补识图，两套方案可直接照做
程序员天天困
1825 13
安全 机器人 API
665 2
缓存 人工智能 算法
758 1
云存储

技术博文

热门文章

最新文章

  • 1
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！
  • 2
    【推荐几款实用的网盘资源搜索引擎】
  • 3
    银行卡归属地及开户行查询API查询实战指南
  • 4
    阿里云盘企业版 CDE Agent 正式发布！企业网盘会思考、能办事，存储即智能
  • 5
  • 6
    揭秘千问 APP 千万级 AI 订单背后的记忆存储实践
  • 7
    阿里云知识存储 Skill 上架阿里云官网首批 Agent Skill：让智能体拥有企业级知识库
  • 8
    让 Agent 拥有记忆 —— 表格存储记忆服务邀测指南
  • 9
    企业AI知识库未来发展趋势深度解读
  • 10
    阿里云 OSS 向量 Bucket 正式商业化，提升 AI 应用效能
  • 1
    阿里云盘企业版 x Qoder 联合发布
    204
  • 2
    阿里云盘企业版 CDE Agent 正式发布！企业网盘会思考、能办事，存储即智能
    297
  • 3
    AI大模型重塑企业知识底座：从RAG到知识智能的技术全景解读
    211
  • 4
    企业AI知识库未来发展趋势深度解读
    180
  • 5
    从RAG到知识图谱：企业AI知识库的技术范式革命与未来十年路线图
    232
  • 6
    AutoMQ x 阿里云 OSS Tables：基于 Iceberg 构建流表一体的实时入湖
    164
  • 7
    指纹浏览器性能横评：100个窗口同时跑，谁的内存和延迟表现最好？
    192
  • 8
    具身智能爆发背后，存储如何成为关键基础设施？
    505
  • 9
    800 家门店的便利零售如何低成本接入 AI 推荐？美好超市基于 OSS 向量 Bucket 的实践
    272
  • 10
    3 人团队零推广获 1.2 万用户：Matrees 如何用 OSS 向量 Bucket 低成本构建 AI 创作平台
    327

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    一条命令迁移，帮你实现 OpenClaw 与 Hermes Agent 记忆互通！