数据工程数据集成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点击查看免费下载3PL Central 是一家提供仓储管理系统WMS服务的 SaaS 供应商其开放 API 覆盖客户、商品、库存、订单与收货等核心仓储业务数据。Airbyte 的source-tplcentral连接器位于 airbyte-integrations/connectors/source-tplcentral正是围绕该 API 构建的数据源连接器每个 Stream 对应一个 API 资源再统一施加四项标准化转换从而把 REST 响应加工成可直接落入数据仓库的规范记录。本文将以连接器自带的 bootstrap.md 为骨架结合streams.py、source.py、util.py、spec.json及单元测试源码逐层拆解其数据模型、转换规则、认证流程与增量同步实现帮助读者理解如何把一款以 C# 模型为 Schema 来源的第三方 API 接入 Airbyte。连接器概览一个 Stream 即一个 API 资源3PL Central 连接器共暴露 6 个数据流全部继承自统一基类TplcentralStream定义于 streams.py其中 4 个支持增量同步Stream 名称对应 API 资源支持同步模式上游主键字段上游游标字段customers/rels/customers/customersfull_refreshReadOnly.CustomerId—items/rels/customers/itemsfull_refresh / incrementalItemIdReadOnly.LastModifiedDateorders/rels/orders/ordersfull_refresh / incrementalReadOnly.OrderIdReadOnly.LastModifiedDatestock_details/rels/inventory/stockdetailsfull_refresh / incrementalReceiveItemIdReceivedDateinventory/rels/inventory/inventoryfull_refresh / incrementalReceiveItemIdReceivedDatestock_summaries/rels/inventory/stocksummariesfull_refresh复合主键见下文—从源码可以看到这 6 个 Stream 由SourceTplcentral.streams()统一实例化并返回source.py每个 Stream 通过重写path()指向不同的 REST 端点。上述配置与 integration_tests/catalog.json 中声明的source_defined_primary_key、supported_sync_modes、default_cursor_field完全一致。四步标准化转换让异构 API 输出统一的记录结构bootstrap.md 明确说明每个 Stream 的响应在进入目标端之前都会依次经历四步转换。这四步并非文档中的概念设计而是实打实编码在基类parse_response与工具函数normalize中streams.py、util.py。1. 字段名从TitleCase规范化为snake_case3PL Central API 的字段是典型的 C# 命名风格如CustomerIdentifier、LastModifiedDate。连接器通过 Airbyte CDK 的规范化机制catalog.json中的字段定义将其统一为snake_case例如ReadOnly.CustomerId对应输出字段_idFacilityId对应facility_id。这一转换保证了不同来源的数据在目标数仓中保持一致的字段命名约定。2. 剔除 HAL_links字段3PL Central API 采用 HALHypertext Application Languagedef _normalizer(dictionary): out {} for key, val in dictionary.items(): if not key _links: out[key] val return out而deep_map会递归处理嵌套的 dict 与 list因此嵌套在任意深度的_links都会被一并清除而不是只清理顶层。3. 为增量 Stream 添加_cursor字段每个增量 Stream 都会在记录上新增一个_cursor字段其值是从上游响应中深层拷贝出来的实际游标字段streams.pydef parse_response(self, response: requests.Response, **kwargs) - Iterable[Mapping]: records normalize(response.json()[self.collection_field]) for record in records: if self.upstream_primary_key: record[self.primary_key] deep_get(record, self.upstream_primary_key) if self.upstream_cursor_field: record[self.cursor_field] deep_get(record, self.upstream_cursor_field) yield record关键在于deep_getutil.py按点分路径逐层取嵌套值——例如Items流的游标字段ReadOnly.LastModifiedDate是两层嵌套结构deep_get会依次取record[ReadOnly][LastModifiedDate]。这样上游字段无论嵌套多深在输出记录中都会被提升为顶层_cursor方便下游按统一字段进行排序与过滤。4. 添加_id或_{name}_id主键字段与_cursor同理每个 Stream 都会新增一个整数型主键字段取值也来自上游深层嵌套的 ID_id当资源自身拥有真实 ID 时使用作为该 Stream 的主键。例如customers使用ReadOnly.CustomerId、orders使用ReadOnly.OrderId、items使用ItemId、stock_details与inventory使用ReceiveItemId。基类中primary_key _id是默认值streams.py。_{name}_id当资源本身没有独立 ID 时使用取依赖对象的 ID 作为组合主键的一部分{name}即该依赖对象的名称。最典型的是stock_summaries流——它没有自己的 ID其主键由FacilityId与_item_identifier_id组合而成streams.pyclass StockSummaries(TplcentralStream): collection_field Summaries primary_key [FacilityId, _item_identifier_id] page_size 500 def parse_response(self, response: requests.Response, **kwargs) - Iterable[Mapping]: records super().parse_response(response, **kwargs) for record in records: record[_item_identifier_id] deep_get(record, ItemIdentifier.Id) yield recordItemIdentifier.Id同样嵌套在对象内部因此在parse_response中被单独提取为顶层_item_identifier_id与顶层FacilityId共同构成复合主键。这一设计也体现在 catalog.json 中stock_summaries的source_defined_primary_key: [[facility_id], [_item_identifier_id]]。转换结果的单元测试验证这四步转换均有对应的单元测试佐证。例如 test_streams.py 中的test_parse_response验证了输入{Foo: foo, Bar: {Baz: baz}, _links: []}会被规范化为{Bar: {Baz: baz}, Foo: foo}_links被移除test_parse_response_with_primary_key验证嵌套的Nested.PrimaryKey会被提取为顶层test_primary_keytest_parse_response_with_cursor_field验证嵌套的Nested.Cursor会被提取为顶层test_cursor_field。Schema 与 C# 模型字段结构以官方 API 文档为准bootstrap.md 特别强调所有 Schema、字段名、结构与注释都对应 3PL Central API 文档https://api.3plcentral.com/Rels/中描述的 C# 模型。这决定了 Schema 文件的设计风格——目录结构与 C# 模型的组织一一对应。在 source_tplcentral/schemas 目录下可以看到清晰的分类层级shared/customer/models/客户域模型item_read_only.json、shipping.json、receiving.json、options.json等 30 个模型文件shared/generic/models/通用模型customer_identifier.json、facility_identifier.json、item_identifier.json、contact_info.json、dimension.json等shared/order/models/订单域模型order_item.json、order_read_only.json、routing_info.json、allocation.json等shared/common/enum/枚举定义address_status_type.json、contact_type.json、warehouse_transaction_api_status.json等顶层 6 个 Schemacustomers.json、items.json、orders.json、inventory.json、stock_details.json、stock_summaries.json则通过$ref引用这些共享模型。以 items.json 为例它除了声明_id与_cursor两个连接器注入字段外其余字段Sku、Upc、Description、Cost、Price、ClassificationIdentifier等与 C# 的 Item 模型一一对应。bootstrap.md 同时指出一个现实问题API 文档有些过时部分端点会返回文档未记载的额外字段。连接器的处理策略是把这些字段补进 Schema 以匹配真实数据——即 Schema 以实际响应为准而不是机械照抄文档。这也是该连接器维护中需要持续关注的差异点。认证机制client_credentials 用户身份标识bootstrap.md 指出API 认证端点要求提供用户登录 IDuser login ID或登录名user login或两者都提供API 凭据可通过服务 UI 获取。连接器通过自定义的TplcentralAuthenticator实现认证source.py。它继承 Airbyte CDK 的Oauth2Authenticator但做了两处关键定制不使用 refresh_token改用client_credentials授权模式。get_refresh_request_body构造的请求体为payload { grant_type: client_credentials, } if self.scopes: payload[scopes] self.scopes if self.user_login_id: payload[user_login_id] self.user_login_id if self.user_login: payload[user_login] self.user_login这正对应 bootstrap.md 中认证端点要求用户登录 ID 或登录名的描述——user_login_id与user_login都作为可选参数拼入令牌请求。使用 HTTP Basic 认证发送 client_id / client_secretresponse requests.post( self.token_refresh_endpoint, authHTTPBasicAuth(self.client_id, self.client_secret), jsonself.get_refresh_request_body(), )check_connection的实现也印证了认证的核心地位它只做一件事——调用get_auth_header()尝试获取令牌成功即返回(True, None)失败则返回错误信息source.py。因此连接配置中最关键的三项凭据是client_id、client_secret以及至少一个用户标识。Customer ID 与 Facility ID贯穿 URL 与查询条件的核心维度bootstrap.md 强调Customer ID 和 Facility ID 会被用于 URL 中无论是路径部分还是查询部分。这在各 Stream 的实现中体现得淋漓尽致且有三种不同的注入方式方式一作为路径参数。Items流的端点路径直接内嵌客户 IDstreams.pydef path(self, **kwargs) - str: return fcustomers/{self.customer_id}/items方式二作为查询参数。StockDetails流把customerid与facilityid作为 URL 查询参数传入streams.py。方式三作为 RQL 过滤条件。Inventory与Orders流把二者编码进 RQLResource Query Language表达式与游标过滤条件通过分号拼接streams.pyparams.update( { sort: self.upstream_cursor_field, rql: ;.join( [ fCustomerIdentifier.Id{self.customer_id}, fFacilityIdentifier.Id{self.facility_id}, ] ), } )Orders 流的 RQL 路径前缀是ReadOnly.因为订单对象的客户/设施标识位于只读嵌套结构中ReadOnly.CustomerIdentifier.Id与ReadOnly.FacilityIdentifier.Id。这两个 ID 在连接配置中由用户显式提供见下节同时customer_id与facility_id也是TplcentralStream.__init__从 config 中读取的核心参数streams.py。分页实现TotalResults pgsiz/pgnum3PL Central API 使用基于TotalResults、pgsiz页大小、pgnum页码的翻页机制。基类TplcentralStream.next_page_token实现了这一逻辑streams.pydef next_page_token(self, response: requests.Response, **kwargs) - Optional[Mapping[str, Any]]: data response.json() total data[self.total_results_field] # 取 TotalResults pgsiz self.page_size or len(data[self.collection_field]) # 显式页大小或按本页实际记录数推断 url urlparse(response.request.url) qs dict(parse_qsl(url.query)) pgsiz int(qs.get(pgsiz, pgsiz)) # 优先读回 URL 中已有的 pgsiz pgnum int(qs.get(pgnum, 1)) # 当前页码默认 1 if pgsiz 0 and pgsiz * pgnum total: # 还有更多数据 return {pgsiz: pgsiz, pgnum: pgnum 1}返回的next_page_token会直接作为下一次请求的查询参数request_params返回next_page_tokenstreams.py。各 Stream 显式设置了不同的页大小Streampage_sizecustomers100items100stock_details500stock_summaries500inventory1000orders1000test_streams.py 中的test_next_page_token系列用例覆盖了边界行为TotalResults为 0 时返回None无下一页当前页记录数为 0 但TotalResults大于 0 时也返回None防御异常响应未显式设置页大小时按本页实际记录数推断分页。增量同步实现游标提取 RQL 过滤 状态合并4 个增量 Stream 继承自IncrementalTplcentralStreamstreams.py其增量能力由三个环节协同完成① 游标字段与状态合并。基类将cursor_field固定为_cursor并通过get_updated_state取当前状态与最新记录的较大值写入 state使用arrow解析时间并统一为无时区的 ISO 格式保证可比性def get_updated_state(self, current_stream_state, latest_record): current current_stream_state.get(self.cursor_field, ) latest latest_record.get(self.cursor_field, ) if current and latest: return {self.cursor_field: max(arrow.get(latest), arrow.get(current)).datetime.replace(tzinfoNone).isoformat()} return {self.cursor_field: max(latest, current)}② 起始位置与切片。stream_slices将每次同步的起点定为已有 state 中的游标值否则为start_datestreams.pyreturn [{self.cursor_field: stream_state.get(self.cursor_field, self.start_date)}]③ 服务端过滤。各 Stream 的request_params会把游标条件编码为 RQL 表达式并加上sort参数按游标字段排序。以Items为例streams.pyparams.update({sort: self.upstream_cursor_field}) cursor stream_slice.get(self.cursor_field) if cursor: params.update({rql: f{self.upstream_cursor_field}ge{cursor}})ge即大于等于配合sort升序排列保证只拉取游标之后的变更数据。注意增量 Stream 在拼接 RQL 时先加入客户/设施过滤再追加游标过滤如Inventory与Orders形成范围过滤 维度过滤的组合查询。此外IncrementalTplcentralStream还设置了state_checkpoint_interval 100即每处理 100 条记录写入一次 checkpoint降低中途失败时的重拉成本。test_incremental_streams.py 验证了这一整套行为test_get_updated_state覆盖了无历史状态无新记录新记录更新游标三种场景test_stream_slices确认无 state 时以start_date作为起始切片test_stream_checkpoint_interval确认 checkpoint 间隔为 100。连接配置参数说明连接器的全部配置项定义在 spec.json 中必填项为url_base、client_id、client_secret。完整参数如下参数类型必填说明url_basestring (uri)是API 基础地址默认https://secure-wms.com/必须以https://开头client_idstring是API 客户端 ID通过 3PL Central 服务 UI 获取client_secretstring是API 客户端密钥airbyte_secret: true界面输入时会脱敏user_login_idinteger否用户登录 ID与user_login至少提供一个认证要求user_loginstring否用户登录名与user_login_id至少提供一个tpl_keystring否3PL GUID示例配置中为{00000000-0000-0000-0000-000000000000}形式的 GUIDcustomer_idinteger否客户 ID用于路径、查询参数与 RQL 过滤facility_idinteger否设施 ID用途同上start_datestring (date-time)否首次增量同步的起始时间RFC 3339 格式如2018-11-13T20:20:3900:00integration_tests/sample_config.json 给出了完整的示例配置骨架其中start_date为2021-10-01是增量 Stream 无历史 state 时的默认起点。小结source-tplcentral连接器是以 API 文档为 Schema 基准、以统一转换保证输出一致、以 RQL 实现服务端增量过滤的典型实现6 个 Stream 直接映射 3PL Central 的 REST 资源TitleCase → snake_case、HAL_links剔除、_cursor游标提升、_id/_{name}_id主键注入四步转换在 util.py 与基类parse_response中落地认证基于client_credentials并强制携带用户身份标识Customer ID / Facility ID 以路径、查询参数、RQL 三种形态贯穿所有请求分页与增量逻辑则分别由TotalResults翻页和 RQLge游标过滤驱动并有完整的单元测试覆盖。对于需要将 3PL Central 仓储数据同步到数仓的工程实践这份连接器源码本身即是可复用的参考蓝本。赞分享数据工程数据集成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 Chargedesk Source 连接器深度解析声明式配置、分页分页与增量同步的实现细节Airbyte Chargedesk Source 连接器深度解析声明式配置、分页分页与增量同步的实现细节 ChargeDesk 是一个聚合多支付网关Str数据工程数据集成ETL后端大数据PostHog Close CRM 数据源连接器全解析从 API 清单到增量同步的实现细节PostHog Close CRM 数据源连接器全解析从 API 清单到增量同步的实现细节 Close CRM 是 PostHog 数据仓库Warehous数据分析后端前端数据可视化大数据Airbyte Google Ads 连接器增量同步原理与实践IncrementalGoogleAdsStream 深度解析Airbyte Google Ads 连接器增量同步原理与实践IncrementalGoogleAdsStream 深度解析 导读 本文聚焦 Airbyte数据工程数据集成ETL后端大数据创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考