跳到主要内容
版本：Next

Airtable

Airtable Sink 连接器

描述​

用于将数据写入 Airtable。

关键特性​

Sink 选项​

参数名类型必须默认值描述
tokenString是-Airtable 个人访问令牌。可在 https://airtable.com/create/tokens 创建。连接器会将其作为 Authorization: Bearer <token> 发送。
base_idString是-Airtable Base ID（以 app 开头）。
tableString是-要写入的表名或表 ID。
api_base_urlString否https://api.airtable.comAirtable API 基础 URL，连接器会自动追加 /v0/<base_id>/<table>。
typecastboolean否false如果为 true，Airtable 会自动将值转换为匹配的字段类型，默认 false。
batch_sizeint否10每次 API 请求的记录数，受 Airtable API 限制最大为 10，默认 10。
request_interval_msint否220API 请求之间的最小间隔（毫秒），默认 220ms（以保持在 Airtable 每秒 5 次请求的限制内），必须 >= 0。
rate_limit_backoff_msint否30000收到 429（限流）响应时的基础退避时间（毫秒），默认 30000ms，必须 >= 0。
rate_limit_max_retriesint否3收到 429 响应后的最大重试次数，默认 3，必须 >= 0。
common-options否-Sink 插件通用参数，详见 Sink Common Options。

使用提示​

  • token 是敏感信息，请避免在共享的任务文件中硬编码真实令牌。可使用 SeaTunnel 变量替换或部署平台的密钥管理机制。
  • 连接器只会写入到固定的 base_id 和 table，不会按上游表名将记录路由到不同的 Airtable 表。多表写入场景需要配置多个 sink，或在上游提前路由。
  • 每条输入记录会变成一条 Airtable 记录。上游字段名需要与 Airtable 列名一致，否则可以开启 typecast = true 让 Airtable 自动转换。
  • Airtable 对每个令牌限速为 5 次/秒。默认 request_interval_ms = 220 可保证单个连接器不超过该限制。可通过 rate_limit_backoff_ms 和 rate_limit_max_retries 控制遇到 HTTP 429 时的退避与重试行为。
  • 连接器不基于时间定时刷新，缓冲达到 batch_size 或写入器关闭时即会发送。

任务示例​

写入 Airtable 表​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        Name = string
        Age = int
      }
    }
    rows = [
      {
        kind = INSERT
        fields = ["Alice", 30]
      },
      {
        kind = INSERT
        fields = ["Bob", 25]
      }
    ]
  }
}

sink {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    typecast = true
    batch_size = 10
    request_interval_ms = 220
  }
}

按列名映射字段写入​

当上游字段名与 Airtable 列名一致时可直接写入：

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        Name = string
        Email = string
        Score = int
      }
    }
    rows = [
      {
        kind = INSERT
        fields = ["Alice", "alice@example.com", 95]
      },
      {
        kind = INSERT
        fields = ["Bob", "bob@example.com", 88]
      }
    ]
  }
}

sink {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Contacts"
    typecast = false
    batch_size = 10
  }
}

写入自托管 Airtable​

可通过 api_base_url 指向自托管或代理的 Airtable 兼容接口：

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        Name = string
        Age = int
      }
    }
    rows = [
      {
        kind = INSERT
        fields = ["Alice", 30]
      }
    ]
  }
}

sink {
  Airtable {
    api_base_url = "https://airtable.internal.example.com"
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
  }
}

变更日志​

Change Log
ChangeCommitVersion