跳到主要内容
版本：3.0.0

Airtable

Airtable 源连接器

描述​

用于从 Airtable 读取数据。

主要特性​

源选项​

参数名类型必须默认值描述
tokenString是-Airtable 个人访问令牌。可在 https://airtable.com/create/tokens 创建。
base_idString是-Airtable Base ID（以 app 开头）。
tableString是-要读取的表名或表 ID。
api_base_urlString否https://api.airtable.comAirtable API 基础 URL。
viewString否-视图名称或 ID，仅返回该视图中可见的记录。
fieldsList否-要包含在响应中的字段名列表。
filter_by_formulaString否-Airtable 公式表达式，用于过滤记录。参考 Airtable 公式文档。
max_recordsint否-返回的最大记录总数。
page_sizeint否-每页记录数（1-100）。
sortString否-排序定义 JSON 数组，例如 [{"field":"Name","direction":"asc"}]。
cell_formatString否-单元格值格式，json 或 string。
return_fields_by_field_idboolean否-如果为 true，响应中的字段键将使用字段 ID 而非字段名。
record_metadataList否-要返回的额外记录元数据，例如 ["commentCount"]。
time_zoneString否-用于格式化日期/时间值的时区。
user_localeString否-用于格式化值的用户区域设置。
offsetString否-Airtable 返回的分页偏移量。通常不需要手动配置，连接器会自动继续读取 Airtable 后续分页。
headersMap否-额外的 HTTP 请求头。连接器会自动添加 Airtable 认证头和 JSON 内容类型请求头。
bodyString否-高级请求体配置。不要把它和 fields、filter_by_formula、page_size、sort 等专用 Airtable 请求选项配置成同一个 Airtable API 字段。
pageingConfig否-继承自 HTTP 连接器的分页配置。普通 Airtable 读取建议优先使用连接器内置的 Airtable 分页处理。
request_interval_msint否220API 请求之间的最小间隔（毫秒），默认 220ms（以保持在 Airtable 每秒 5 次请求的限制内）。
rate_limit_backoff_msint否30000收到 429（限流）响应时的基础退避时间（毫秒），默认 30000ms。
rate_limit_max_retriesint否3收到 429 响应后的最大重试次数，默认 3。
schemaConfig否-输出数据结构，format = "json" 时必填。详见 Schema 特性。
schema.fieldsConfig否-字段名与 SeaTunnel 数据类型，用于解析 JSON 响应。
formatString否text上游数据的格式，支持 json 和 text，默认 text。
content_fieldString否-用于从响应中提取数据的 JsonPath 表达式。对于 Airtable，通常使用 $.records[*].fields 来提取每条记录的字段。
json_fieldConfig否-字段级 JSONPath 映射，与 schema 配合使用。
json_filed_missed_return_nullboolean否false配置的 JSON 字段缺失时是否返回 null，否则报错。
enable_multi_linesboolean否false是否启用多行模式，将响应体中按换行分隔的多个 JSON 对象视为独立记录。
connect_timeout_msint否12000HTTP 连接超时时间（毫秒），默认 12000ms。
socket_timeout_msint否60000HTTP 套接字超时时间（毫秒），默认 60000ms。
common-optionsconfig否-源插件通用参数，详见 源通用选项。

使用提示​

  • token 是敏感信息，请避免在共享的任务文件中硬编码真实令牌。可使用 SeaTunnel 变量替换或部署平台的密钥管理机制。
  • 连接器会自动按 Airtable offset 分页继续读取，通常不需要手动设置 offset。
  • Airtable 对每个令牌限速为 5 次/秒。默认 request_interval_ms = 220 可保证单个连接器不超过该限制。可通过 rate_limit_backoff_ms 和 rate_limit_max_retries 控制遇到 HTTP 429 时的退避与重试行为。
  • 若希望输出带字段名的数据行，请将 format 设置为 json 并配置 schema。
  • 使用 content_field = "$.records[*].fields" 在解析前先抽取记录的字段部分。
  • 只有当不同字段位于不同 JSON 路径时，才需要使用 json_field。

任务示例​

读取记录并以文本形式输出​

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

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    format = "text"
    max_records = 10
  }
}

sink {
  Console {
  }
}

指定 Schema 并提取字段​

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

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    format = "json"
    content_field = "$.records[*].fields"
    filter_by_formula = "{Status} = 'Shipped'"
    schema = {
      fields {
        Name = string
        Status = string
        Weight = float
      }
    }
  }
}

按页读取并控制吞吐​

通过 page_size 与 request_interval_ms 控制分页读取的吞吐：

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

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    format = "json"
    content_field = "$.records[*].fields"
    page_size = 2
    request_interval_ms = 220
    schema = {
      fields {
        Name = string
        Age = int
        Status = string
      }
    }
  }
}

通过 JSONPath 抽取字段​

当不同字段位于不同 JSON 路径时，使用 json_field：

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    format = "json"
    content_field = "$.records[*]"
    json_field = {
      Name = "$.fields.Name"
      Status = "$.fields.Status"
      CreatedAt = "$.createdTime"
    }
    schema = {
      fields {
        Name = string
        Status = string
        CreatedAt = string
      }
    }
  }
}

仅读取某个视图中的记录​

使用 view 只读取指定视图中可见的记录，配合 fields 限制返回的列：

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    view = "Pending shipments"
    fields = ["Name", "Status", "Weight"]
    format = "json"
    content_field = "$.records[*].fields"
    schema = {
      fields {
        Name = string
        Status = string
        Weight = float
      }
    }
  }
}

按增量批次消费新增记录​

Airtable source 只支持 BATCH 作业（其它模式在初始化时会被拒绝）。要在多次运行之间持续 消费新增行，可以固定 filter_by_formula 并配合 sort 按 createdTime 升序排列，同时把 上一次运行读到的最大 CreatedAt 写回公式：

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

source {
  Airtable {
    token = "patXXXXXXXX.XXXXXXXX"
    base_id = "appXXXXXXXX"
    table = "Shipments"
    format = "json"
    content_field = "$.records[*].fields"
    filter_by_formula = "IS_AFTER({CreatedAt}, '2026-01-01T00:00:00.000Z')"
    sort = "[{\"field\":\"CreatedAt\",\"direction\":\"asc\"}]"
    page_size = 100
    schema = {
      fields {
        Name = string
        Status = string
        CreatedAt = string
      }
    }
  }
}

变更日志​

Change Log
ChangeCommitVersion
[Improve][Connector-V2][Airtable] Add jitter to rate-limit backoff (#11654)https://github.com/apache/seatunnel/commit/7e24d0f423.0.0
[Feature][Connector-V2] Add Airtable source and sink (#10469)https://github.com/apache/seatunnel/commit/c15032c033.0.0