数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载本文基于 Mage 开源仓库中的 Airtable 目标端Destination文档系统讲解如何把数据管道产出的记录写入 Airtable包括token与base_id两个必填配置项、Personal Access Token 的生成方式与所需权限范围并深入源码层面剖析建表、类型映射、批量写入的完整实现链路。读完本文你可以独立在 Mage 中配置一个可用的 Airtable 数据同步目标并理解每条记录被写入前经历了哪些处理步骤。配置参数Airtable 目标端属于 Mage 的 data integrations 目标端体系位于仓库mage_integrations/mage_integrations/destinations/airtable/。要启用该目标端必须在目标端配置中提供以下两个参数Key说明示例值是否必填token拥有访问目标表权限的 Airtable 访问令牌Personal Access Tokenpat...✅ 必填base_idAirtable 工作区中指定应用Base的 ID通常以app开头app...✅ 必填配置骨架可直接参考目标端的模板文件 templates/config.json内容即这两个字段的占位形式{ token: , base_id: }在 Mage 的可视化界面中创建 Data Integration Pipeline 时选择 Airtable 作为目标端即可在弹窗中填写上述字段在命令行模式下则通过--config_json传入 JSON 或通过--config指定配置文件路径参数解析见 destinations/base.py。其中base_id是 Airtable Base 的唯一标识可从该 Base 的浏览器 URL 中获取形如https://airtable.com/appXXXXXXXXXXX/...也可通过 Airtable API 查询得到。一个访问令牌可以同时操作多个 Base但每个目标端实例只绑定一个base_id。生成访问凭据Airtable 目标端要求使用Personal Access Token个人访问令牌而非旧的 API Key。生成步骤概括如下登录 Airtable 并进入开发者后台的 Personal Access Token 创建页面点击Create token新建令牌在创建向导中为令牌选择所需的权限范围Scopes见下文以及可访问的 Base生成后立即安全保存令牌令牌只在创建时完整展示一次。生成的令牌以pat开头与上文中配置表的示例值一致。令牌所需权限范围为让目标端能创建表和写入记录你的令牌必须包含以下两个 Scopes写入目标端要求与读取类 Source 不同——Airtable 源端 sources/airtable/README.md 要求的则是data.records:read与schema.bases:readschema.bases:write—— 允许读取与修改 Base 的结构目标端创建表时依赖该权限data.records:write—— 允许对 Airtable 中的记录执行插入、更新、删除目标端批量写入记录时依赖该权限。缺少任何一个权限连接测试或写入阶段都会抛出 Airtable API 权限错误。源码链路从连接测试到建表写入连接构建与连接测试目标端入口类是 destinations/airtable/init.py 中的Airtable它继承自mage_integrations.destinations.base.Destination。build_client()从配置中取出token与base_id实例化 connections/airtable/init.py 中的AirtableConnectionconnection AirtableConnection(self.config[token], self.base_id) return connection.build_connection()底层连接封装了pyairtable库AirtableConnection.__init__用Api(self.token)创建 API 客户端build_connection()返回self.api.base(self.base_id)即指定的 Base 对象。整个目标端没有单独维护数据库连接状态而是在每次调用测试连接、批量导出时重建客户端。test_connection()通过client.tables()拉取 Base 下的表列表来验证令牌与base_id是否有效成功则记录Airtable connection test successful.失败则抛出异常。这一方法会被Destination.process()中的--test_connection模式触发。建表与 JSON 类型映射export_batch_data()是每次批量写入的核心。对于当前 stream它首先依据 stream schema 的properties生成 Airtable 字段列表fields [ { name: column, type: map_json_to_airtable(info[type]), **({options: {precision: 2}} if map_json_to_airtable(info[type]) number else {}) } for column, info in self.schemas[stream][properties].items() ] table client.create_table(stream, fields)其中map_json_to_airtable定义在 mage_ai/io/utils.py负责把 Mage 管道的 JSON Schema 类型映射为 Airtable 字段类型JSON Schema 类型Airtable 字段类型stringmultilineTextintegernumbernumbernumber并设置options.precision 2booleancheckboxarraymultipleSelectsobjectsingleCollaboratordate-timedateTime其他/缺省str映射逻辑会先剔除类型列表中的null因为 Mage schema 中字段类型常写成[null, string]形式再取第一个非空类型进行匹配。create_table成功后会记录Successfully created table {table.name}。内部列与批量写入写入前目标端会给每条记录追加两个内部审计列定义于 destinations/constants.py_mage_created_at_mage_updated_atupdate_record_with_internal_columns见 destinations/utils.py以 UTC 时间戳填充这两列def update_record_with_internal_columns(record): curr_time datetime.utcnow().strftime(%Y-%m-%d %H:%M:%S.%f) record[INTERNAL_COLUMN_CREATED_AT] curr_time record[INTERNAL_COLUMN_UPDATED_AT] curr_time return record随后目标端把记录组装成records [item[record] for item in record_data]调用table.batch_create(records)一次性写入 Airtable并通过日志 tagsrecords、stream、records_inserted记录批量大小与成功条数。命令行运行与批量处理机制Airtable类支持直接作为命令行程序运行文件末尾的__main__分支destination Airtable( argument_parserargparse.ArgumentParser(), batch_processingTrue, ) destination.process(sys.stdin.buffer)它会从标准输入读取 singer 格式的 JSON Lines 流RECORD、SCHEMA、STATE等消息由基类Destination.process()统一调度。基类支持的命令行参数包括--config_json直接传入配置 JSON--config指定配置文件路径--catalog_json传入 catalog JSON预置 stream 及 schema--test_connection仅测试连接不写入--show_templates输出目标端配置模板--state指定状态文件路径--input_file_path、--debug、--log_to_stdout等辅助参数。基类默认以批量模式处理数据见 destinations/base.py记录先按 stream 累积当累计字节数达到maximum_batch_size_mb默认 100 MBMAXIMUM_BATCH_SIZE_MB 100时触发一批写入并在批次之间用batch_number标记日志。因此目标端实际是按批次调用export_batch_data()而非逐条写入这对 Airtable API 的调用频率与吞吐量控制是有利的。批次处理完成后基类还会合并各 stream 的 bookmarks 并写出最终 state。常见问题base_id从哪来它是 Airtable Base 的唯一标识通常以app开头可在 Base 的 URL 中查看也可通过 Airtable API 获取。令牌如何安全存放建议使用环境变量或 Mage 的 Secrets Manager 存储token避免把令牌硬编码进配置文件或提交到版本控制。能否写入多个 Airtable Base可以但每个目标端实例对应一个base_id写入不同 Base 需分别配置目标端并确保令牌对该 Base 拥有相应权限。与 Airtable 源端配置有何区别源端额外需要table_nametbl...参数且只需只读权限目标端则通过base_id定位 Base并按 stream 自动建表权限要求为写入类。本文对应的完整目标端文档亦收录在仓库文档站 docs/data-integrations/destinations/airtable.mdx 中可与源码对照阅读。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐使用 Mage 的 Airtable Source 连接器同步无代码数据库数据使用 Mage 的 Airtable Source 连接器同步无代码数据库数据 导读 Airtable 是一款流行的“无代码数据库”产品用户经常需要把其中的表数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage-ai Chargebee 数据源接入指南配置、Stream 同步机制与源码实现解析Mage ai Chargebee 数据源接入指南配置、Stream 同步机制与源码实现解析 Chargebee 是 SaaS 订阅计费平台在 mage a数据工程数据编排ETL任务调度批处理流处理数据集成后端前端mage-ai Freshdesk 数据源接入指南配置参数、增量同步机制与源码实现解析mage ai Freshdesk 数据源接入指南配置参数、增量同步机制与源码实现解析 本文以开源仓库 mage ai 中 Freshdesk 数据源 htt数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇【亲测免费】 探索Vortex一款强大的GPGPU编程框架下一篇探索神秘代码库MakeItTalk - 让你的程序开口说话创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考