Airbyte 声明式连接器实践：source-castor-edc 的 Low-Code Manifest 架构、配置与本地开发指南

数据工程数据集成ETL后端大数据【免费下载链接】airbyteOpen-source data movement for ELT pipelines and AI agents — from APIs, databases files to warehouses, lakes, and AI applications. Both self-hosted and Cloud.项目地址https://gitcode.com/gh_mirrors/ai/airbyte点击查看免费下载本指南以source-castor-edcCastor EDC 临床研究数据源连接器为核心系统讲解 Airbyte 声明式连接器Declarative / Low-Code CDK的完整形态从 Connector Builder 生成的manifest.yaml如何定义认证、数据流、分页与增量同步到四个必填配置项的实际含义再到单元测试与验收测试的组织方式。读完本文你将掌握如何阅读、配置、测试与本地运行一个 manifest-only 连接器并能将同一套模式迁移到其他声明式连接器上。一、连接器概览一个由 Connector Builder 生成的声明式连接器source-castor-edc 的 README 开宗明义地指出这是一个用 Connector Builder 构建的声明式连接器Declarative Connector。与传统的 Python/Java 手写连接器不同声明式连接器不需要编写逐行的请求、解析与状态管理代码而是通过一份结构化的 YAML 清单manifest声明如何请求、如何解析、如何分页、如何增量由 Airbyte 的 Low-Code CDK声明式 CDK在运行时解释执行。在 metadata.yaml 中可以确认该连接器的技术属性connectorType: source、connectorSubtype: api属于 REST API 类源连接器tags同时标注了language:manifest-only与cdk:low-code表明它没有语言级业务代码全部逻辑由 manifest 承载releaseStage: alpha、supportLevel: communityreleaseDate: 2024-10-12当前dockerImageTag: 0.0.64connectorBuildOptions.baseImage指向airbyte/source-declarative-manifest:7.33.0即运行时代理镜像印证了manifest 驱动 通用运行镜像的运行模式allowedHosts列出了三个区域主机data.castoredc.com、uk.castoredc.com、us.castoredc.com。连接器目录结构非常精简全部文件如下文件/目录作用manifest.yaml连接器的唯一实现声明流、认证、配置、schema约 2500 行metadata.yaml注册元数据定义 ID、镜像名、发布阶段、允许主机等acceptance-test-config.ymlConnector Acceptance TestsCAT配置unit_tests/test_manifest.py针对 manifest 行为的单元测试unit_tests/pyproject.toml测试环境声明airbyte-cdk 7.33.0、pytest ^8.0、pyyaml ^6.0icon.svg连接器图标可见对这种 manifest-only 连接器而言README 只是入口manifest.yaml 才是灵魂。下文将以 manifest.yaml 为主线逐层拆解。二、manifest 顶层骨架check、streams 与 spec 三大件manifest.yaml 的顶层结构是标准声明式连接器三件套version: 5.12.0 type: DeclarativeSource description: Documentation: https://uk.castoredc.com/api#/ check: type: CheckStream stream_names: - user definitions: # 所有可复用的组件定义streams、base_requester streams: # 暴露给 Airbyte 的 16 个数据流引用 definitions 中的流 spec: # 连接器的配置表单connection_specification schemas: # 每个流的内联 JSON Schematype: DeclarativeSource告诉 Low-Code CDK 这是一个声明式连接器version: 5.12.0是 manifest 格式版本号check阶段使用CheckStream通过请求user流来验证连接配置是否有效——这是连接器在建立连接时执行的连通性检查definitions是组件的仓库streams一节通过$ref引用其中的流这种定义-引用分离的结构让多个流可以复用同一个请求器requester。具体到本连接器16 个流在 streams 列表 中逐一注册而schemas一节为每个流提供了内联 JSON SchemaInlineSchemaLoader。三、认证与区域路由base_requester 与 OAuth client_credentials所有流共享同一个请求器base_requester定义在 manifest.yaml 的 definitions 段base_requester: type: HttpRequester url_base: https://{{ data if config[url_region] nl else config[url_region] }}.castoredc.com/api/ authenticator: type: OAuthAuthenticator client_id: {{ config[\client_id\] }} grant_type: client_credentials client_secret: {{ config[\client_secret\] }} refresh_request_body: {} token_refresh_endpoint: https://{{ data if config[url_region] nl else config[url_region] }}.castoredc.com/oauth/token这里体现了两个值得注意的设计区域动态路由url_base使用 Jinja 模板表达式当url_region为nl时主机为data.castoredc.comCastor 荷兰区使用data子域其余情况直接使用区域值uk、us。这与 metadata.yaml 中allowedHosts的三个主机完全对应也呼应了 Castor 官方用户文档 中nl同时用于 API 请求与 OAuth 认证的说明。OAuth client_credentials 流程认证器采用OAuthAuthenticatorgrant_type为client_credentials通过client_id/client_secret向/oauth/token端点换取访问令牌令牌随后自动附加到 API 请求中。refresh_request_body: {}表示刷新请求体为空。从流定义可以看到每个流的requester都通过$ref: #/definitions/base_requester复用该认证逻辑。这一行为并非笔者的主观推断unit_tests/test_manifest.py 提供了机器可验证的证据测试将 manifest 中的base_requester交给ModelToComponentFactory实例化为真实组件然后断言三个区域的 API 地址与令牌端点分别为assert api_url fhttps://{host}/api/ assert token_url fhttps://{host}/oauth/token assert requester.authenticator.get_grant_type() client_credentials其中nl - data.castoredc.com、uk - uk.castoredc.com、us - us.castoredc.com并校验主机均在allowedHosts白名单内。第二项测试 test_region_configuration_remains_compatible 则锁定url_region的枚举值必须为[uk, nl, us]且默认值为uk防止未来改动破坏配置兼容性。四、16 个数据流全解三类流模式Castor 官方用户文档 的 Streams 一节给出了 16 个流的完整清单。结合 manifest.yaml 的实现可以进一步整理出每个流的主键、分页策略、增量能力与记录提取路径流名主键分页增量游标字段记录提取路径DpathExtractor父流userid无支持last_login_embedded.user-studycrf_idDefaultPaginator每页 50支持created_on_embedded.study-audit_trialuuidDefaultPaginator每页 1000支持dateitemsstudycountryid无不支持results-study_surveyidDefaultPaginator不支持_embedded.surveysstudystudy_survey_packageidDefaultPaginator不支持_embedded.survey_packagesstudystudy_siteidDefaultPaginator不支持_embedded.sitesstudystudy_fieldsidDefaultPaginator不支持_embedded.fieldsstudystudy_field_dependencyidDefaultPaginator不支持_embedded.fieldDependenciesstudystudy_field_validationidDefaultPaginator不支持_embedded.fieldValidationsstudystudy_formidDefaultPaginator不支持_embedded.formsstudystudy_roleuuidDefaultPaginator不支持_embedded.rolesstudystudy_fieldoption_groupsidDefaultPaginator不支持_embedded.fieldOptionGroupsstudystudy_statisticsstudy_idDefaultPaginator不支持空路径取整个响应体studystudy_useridDefaultPaginator支持last_login_embedded.studyUsersstudystudy_visitidDefaultPaginator不支持_embedded.visitsstudy从源码结构可以归纳出三类典型的流模式模式一顶层列表流无需父流。user、study、country直接请求根级端点如GET /api/user、GET /api/study其中user与country甚至不配置分页器属于一次性全量拉取study是其余 13 个流的数据源根。模式二按 study 分区的子流Substream。绝大多数流属于此类其端点路径形如study/{{ stream_partition[study_id] }}/site、study/{{ stream_partition[study_id] }}/form并通过SubstreamPartitionRouter将父流study的每条记录拆分为一个分区请求。典型配置以study_visit为例见 manifest.yamlpartition_router: type: SubstreamPartitionRouter parent_stream_configs: - type: ParentStreamConfig parent_key: study_id # 父流 study 中取值字段 partition_field: study_id # 注入子流路径的占位符 stream: $ref: #/definitions/streams/study这意味着连接器先同步study流再为每个study_id依次请求其下的调查问卷survey、表单form、站点site、字段field、角色role等子资源。模式三统计聚合流。study_statistics的record_selector使用空field_path: []从源码结构看这表示不深入响应对象而是将整个响应体作为记录输出用于获取每个研究的记录数与按机构拆分的统计信息。record_selector统一采用RecordSelectorDpathExtractor组合例如user流从_embedded.user提取记录Castor API 遵循 HATEOAS 风格数据包裹在_embedded下country流则从results提取这与各端点实际的响应结构一一对应。五、分页、限流与重试策略除user、country外其余流都配置了DefaultPaginator与PageIncrement分页策略。以study流为例manifest.yamlpaginator: type: DefaultPaginator page_token_option: type: RequestOption inject_into: request_parameter field_name: page page_size_option: type: RequestOption field_name: page_size inject_into: request_parameter pagination_strategy: type: PageIncrement page_size: 50 start_from_page: 1 inject_on_first_request: true要点解析分页参数page与page_size以query 参数request_parameter方式注入请求PageIncrement表示每页页码递增start_from_page: 1从第 1 页开始inject_on_first_request: true表示首次请求也携带页码参数study流每页 50 条而其余子流如audit_trial、study_survey每页 1000 条不同流的page_size可根据接口特性差异化设置。所有流都配置了统一的限流与重试策略CompositeErrorHandler例如 manifest.yamlerror_handler: type: CompositeErrorHandler error_handlers: - type: DefaultErrorHandler max_retries: 5 response_filters: - type: HttpResponseFilter action: RATE_LIMITED http_codes: - 429 error_message: Rate limits has been hit backoff_strategies: - type: ConstantBackoffStrategy backoff_time_in_seconds: 5含义当接口返回 429请求过于频繁时连接器将命中RATE_LIMITED过滤器以每 5 秒固定退避的方式最多重试 5 次。这种识别限流码 固定退避的模式是声明式连接器处理第三方 API 限流的标准做法避免同步任务因瞬时限流而失败。六、增量同步与时间游标DatetimeBasedCursor4 个流支持增量同步user游标last_login、study游标created_on、audit_trial游标date、study_user游标last_login。它们统一使用DatetimeBasedCursor以study流为例manifest.yamlincremental_sync: type: DatetimeBasedCursor cursor_field: created_on cursor_datetime_formats: - %Y-%m-%d %H:%M:%S datetime_format: %Y-%m-%d %H:%M:%S start_datetime: type: MinMaxDatetime datetime: {{ config[\start_date\] }} datetime_format: %Y-%m-%dT%H:%M:%SZ end_datetime: type: MinMaxDatetime datetime: {{ now_utc().strftime(%Y-%m-%dT%H:%M:%SZ) }} datetime_format: %Y-%m-%dT%H:%M:%SZ要点解析游标字段分别为created_on研究创建时间与last_login用户最近登录时间start_datetime取用户配置的start_dateend_datetime取当前 UTC 时间now_utc()构成同步窗口audit_trial与study_user还额外通过start_time_option/end_time_option将窗口映射为请求参数date_from/date_to见 manifest.yaml说明这两个接口以服务端过滤配合本地游标双重实现增量。audit_trial流的transformations部分展示了声明式字段变换的用法manifest.yamltransformations: - type: AddFields fields: - path: [date] value: {{ record[\datetime\][\date\].split(\.\)[0] }} - type: AddFields fields: - path: [uuid] value: {{ now_utc() }}第一条变换从嵌套的datetime.date对象中截取日期字符串作为顶层date字段也是增量游标第二条为每条记录注入now_utc()生成的 UUID充当该流的主键primary_key: uuid。study_role流同样使用now_utc()生成uuid主键。这体现了声明式连接器以变换补足主键与游标的常见技巧。七、配置表单与使用方式四个必填参数连接器的配置表单定义在 manifest.yaml 的 spec 段四个字段全部必填字段类型说明默认值url_regionstring研究数据所在区域uk英国、nl荷兰、us美国ukclient_idstring在对应区域账户设置中生成的 API Client ID-client_secretstring对应区域的 API Client Secret-start_datestring增量同步的起始时间-细节说明client_id与client_secret均标记airbyte_secret: trueAirbyte 平台会将其作为敏感信息加密存储并在 UI 中脱敏显示start_date使用format: date-time并带正则约束^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$即必须为YYYY-MM-DDTHH:MM:SSZ形式的 UTC 时间戳与MinMaxDatetime中%Y-%m-%dT%H:%M:%SZ的解析格式严格对齐区域与认证凭据的获取位置见 Castor 官方用户文档 的 Authentication 一节url_region服务器账户设置入口nl荷兰EU主机data.castoredc.comdata.castoredc.com/account/settingsuk英国主机uk.castoredc.comuk.castoredc.com/account/settingsus美国主机us.castoredc.comus.castoredc.com/account/settings该文档同时提醒若使用 Airbyte Cloud 且组织启用了 IP 白名单需要将 Airbyte Cloud 的出口 IP 加入允许列表确保连接器能够访问上述三个区域主机。八、测试与验收单元测试与 Connector Acceptance Tests单元测试直接实例化 manifest 组件unit_tests/test_manifest.py 是声明式连接器测试的典型范式——不 mock 网络而是把 manifest 解析为真实 CDK 组件并断言其配置正确性。测试代码通过ModelToComponentFactory().create_component(model_typeHttpRequester, ...)将base_requester反序列化为可执行的HttpRequester对象然后验证三个区域的 API 基地址与 OAuth 令牌端点拼接正确认证类型确为client_credentials生成的 URL 主机均在allowedHosts白名单内url_region的枚举与默认值保持不变。依赖声明见 unit_tests/pyproject.tomlairbyte-cdk 7.33.0与运行镜像版本一致、pytest ^8.0、pyyaml ^6.0。验收测试CAT 配置acceptance-test-config.yml 声明了 Connector Acceptance TestsCAT的入口其中spec测试直接指向manifest.yaml即验证 manifest 生成的 spec 符合协议规范而connection、discovery、basic_read、incremental、full_refresh均以bypass_reason: This is a builder contribution, and we do not have secrets at this time跳过——这是因为该连接器由 Connector Builder 社区贡献暂无测试凭据。需要说明的是metadata.yaml的metadata.testedStreams段记录了 16 个流的测试结果标记hasRecords、primaryKeysAreUnique、responsesAreSuccessful等均为 true说明这些流的响应结构已通过构建侧校验。九、本地开发与连接器特定指南README 的 Development 一节指出声明式连接器的本地开发与测试流程遵循 Airbyte 的本地连接器开发指南核心思路是在本地构建镜像后通过spec、check、discover、read等协议命令驱动连接器并配合单元测试与 CAT 验证行为。对 manifest-only 连接器而言本地迭代的主要对象就是 manifest.yaml 本身修改后重新构建airbyte/source-castor-edc:dev镜像即可验证对应 acceptance-test-config.yml 中的connector_image: airbyte/source-castor-edc:dev。README 的 Connector-Specific Guidance 一节还提到连接器特定的排错与测试指导可以记录在连接器目录下的CONTRIBUTING.md中当前仓库快照的该目录下未包含此文件实际以发行产物为准。这提醒连接器维护者把只有本连接器才成立的踩坑经验与通用模板 README 分开存放是 Airbyte 连接器仓库的约定做法。十、总结source-castor-edc是理解 Airbyte 声明式连接器的最佳范例之一它在 manifest.yaml 中浓缩了声明式连接器的几乎所有核心能力——基于client_credentials的 OAuth 认证、通过 Jinja 表达式实现的区域动态路由、SubstreamPartitionRouter的父子流分区同步、PageIncrement分页、429 限流重试、DatetimeBasedCursor增量同步以及AddFields字段变换——并且全部由 单元测试 锁定了关键行为。阅读这类连接器时正确的姿势是以 README 为入口以 manifest 为实现以 metadata 为注册信息以测试为行为契约。当你需要为下一个 REST API 编写连接器时这套声明式三板斧完全可以直接复用。赞分享数据工程数据集成ETL后端大数据【免费下载链接】airbyteOpen-source data movement for ELT pipelines and AI agents — from APIs, databases files to warehouses, lakes, and AI applications. Both self-hosted and Cloud.项目地址https://gitcode.com/gh_mirrors/ai/airbyte点击查看免费下载相关推荐Airbyte Gutendex 声明式 Source 连接器深度解析基于 Low-Code CDK 的 manifest 实现与实战配置Airbyte Gutendex 声明式 Source 连接器深度解析基于 Low Code CDK 的 manifest 实现与实战配置 本篇文章以 Air数据工程数据集成ETL后端大数据Airbyte Employment Hero 声明式连接器深度解析基于 Low-Code CDK 的 manifest 实现与本地开发指南Airbyte Employment Hero 声明式连接器深度解析基于 Low Code CDK 的 manifest 实现与本地开发指南 本篇技术指南以开数据工程数据集成ETL后端大数据Airbyte News API Source 连接器实战指南manifest-only 声明式连接器的架构、配置与测试Airbyte News API Source 连接器实战指南manifest only 声明式连接器的架构、配置与测试 本文以 Airbyte 仓库中的 s数据工程数据集成ETL后端大数据上一篇10倍压缩PP-HumanSeg移动端优化实战从100MB到10MB的极致瘦身下一篇闻达本地部署与云端部署对比如何选择最适合的部署方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考