数据工程数据集成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点击查看免费下载导读Apple Search AdsApple Ads是苹果官方的搜索广告投放平台。本文以 Airbyte 仓库中source-apple-search-ads连接器为核心完整讲解其基于 Airbyte Low-Code CDK 声明式配置实现的数据流设计实体类 Base StreamsCampaigns / AdGroups / Keywords / Ads与统计类 Report Streams四个_daily报告流的同步模式、增量游标机制、OAuth 认证、重试与分页策略以及全部配置参数的取值与含义。读完本文你将掌握该连接器的数据获取边界、同步行为与调优手段能够据此正确配置源并诊断同步问题。一、连接器概览REST API 之上的声明式实现Apple Search Ads 对外提供的是 REST 风格 API。source-apple-search-ads连接器并非手写 Python/Java 实现而是采用Airbyte Low-Code CDK的声明式描述方式所有请求、认证、分页、错误处理与增量逻辑都定义在 manifest.yaml 中由 CDK 运行时source-declarative-manifest镜像解释执行。这一点与仓库中 README.md 描述的 declarative connector built with the Connector Builder 一致其 metadata.yaml 也标注了cdk:low-code与language:manifest-only标签。API 基址定义在base_requester中https://api.searchads.apple.com/api/v5连接器整体将数据流划分为两大类对应 bootstrap.md 的核心脉络类别用途同步模式Base streamsAPI 中的实体属性有哪些 Campaign、AdGroup、Keyword、Ad仅 Full RefreshReport streams实体统计指标Campaign 花了多少钱、Keyword 有多少次点击等Full Refresh Incremental二、Base Streams实体类数据流Base streams 返回 Apple Search Ads 账户中实体的属性信息全部只支持全量刷新full refresh。连接器定义了 4 个campaigns推广计划adgroups广告组keywords关键词ads广告每个流的实体主键都是id整数类型这可以从 configured_catalog.json 中看到四个流均声明source_defined_primary_key: [[id]]与supported_sync_modes: [full_refresh]。在 manifest.yaml 中Base streams 的请求路径体现了 Apple Search Ads API 的层级关系流HTTP路径campaignsGET/campaignsadgroupsGET/campaigns/{{ stream_slice.campaign_id }}/adgroupskeywordsGET/campaigns/{campaign_id}/adgroups/{adgroup_id}/targetingkeywordsadsGET/campaigns/{campaign_id}/adgroups/{adgroup_id}/ads注意后三个流的路径中带有模板变量AdGroups 隶属于某个 CampaignKeywords 与 Ads 又嵌套在 AdGroup 之下。这对应 manifest 中的SubstreamPartitionRouter——adgroups以campaigns.id为父键分区keywords/ads再以adgroups.id为父键二次分区从而实现先拉取 Campaign 列表再逐 Campaign 拉取 AdGroup再逐 AdGroup 拉取 Keyword / Ad的层级遍历。所有 Base stream 请求都会携带组织上下文头X-AP-Context: orgId{{ config.org_id }}即每次调用都必须指明数据所属的组织Org。三、Report Streams统计类数据流与 DAILY 粒度Report streams 返回各实体的统计指标花费、点击、展示等同时支持全量刷新与增量同步。连接器提供 4 个campaigns_report_dailyCampaign 级报告adgroups_report_dailyAd Group 级报告keywords_report_dailyKeyword 级报告ads_report_dailyAd 级报告当前报告流只设置为DAILY粒度即请求体中的granularity: DAILY因此实际数据流名称统一带有_daily后缀。这也是 bootstrap.md 明确指出的现状。从 configured_catalog.json 可以看到报告流的增量特征默认游标字段cursor为date且由源定义source_defined_cursor: true支持full_refresh与incremental两种同步模式复合主键为date 实体 IDcampaigns_report_daily为[date, campaignId]adgroups_report_daily为[date, adGroupId]keywords_report_daily为[date, keywordId]ads_report_daily为[date, adId]目标写入模式为append追加增量同步结果按天持续追加到目标端。3.1 报告接口的调用方式报告流与实体流不同使用的是 POST 请求且分区路由只到 Campaign 层级按campaign_id分区流HTTP路径campaigns_report_dailyPOST/reports/campaignsadgroups_report_dailyPOST/reports/campaigns/{campaign_id}/adgroupskeywords_report_dailyPOST/reports/campaigns/{campaign_id}/keywordsads_report_dailyPOST/reports/campaigns/{campaign_id}/ads每个报告请求的 JSON 请求体request_body_json结构如下request_body_json: startTime: {{ stream_slice.start_time }} endTime: {{ stream_slice.end_time }} granularity: DAILY groupBy: [ countryOrRegion ] selector: { orderBy: [ { field: countryOrRegion, sortOrder: ASCENDING } ] } timeZone: {{ config[timezone] or UTC }}startTime/endTime来自增量同步产生的日期切片slice见下文granularity固定为DAILYgroupBy按国家/地区countryOrRegion分组统计selector中同时按该字段升序排序timeZone决定报告统计的时区口径默认UTC也可配置为ORTZOrganization Time Zone组织时区。报告接口的响应被封装在data.reportingDataResponse.row路径下因此 manifest 中DpathExtractor的field_path为[data, reportingDataResponse, row]分页的 offset/limit 则注入到请求体 JSON 的selector.pagination中每页 1000 条。3.2 报告的字段变换transformationsApple 报告接口返回的原始行数据把指标与元信息混在metadata字段中因此 manifest 通过AddFields变换把关键信息提升为顶层字段例如campaigns_report_dailytransformations: - type: AddFields fields: - type: AddedFieldDefinition path: [campaignId] value: {{ record.metadata.campaignId }} - type: AddedFieldDefinition path: [date] value: {{ stream_slice.start_time }} - type: AddedFieldDefinition path: [countryorregion] value: {{ record.metadata.countryOrRegion }}实体 IDcampaignId/adGroupId/keywordId/adId从record.metadata中提出date直接取当前日期切片起点stream_slice.start_timecountryorregion同样从metadata.countryOrRegion提取。将countryOrRegion单独提取为顶层字段是有实际意义的报告按国家/地区分组后若仍以整个metadata字段作为主键进行去重metadata中其他键的变化会导致同一date 实体 ID 的记录出现重复单独提取后即可用countryOrRegion代替整个metadata参与主键去重。这也是 manifest.yaml 顶部 description 中解释的修复动机。四、增量同步机制start_date、end_date 与日粒度切片报告流支持增量同步其核心是 manifest 中的DatetimeBasedCursor时间游标。bootstrap.md 明确说明连接器使用start_date配置项作为首次报告同步的起点若未显式设置end_date则以当前日期作为结束日期。对应 manifest 的实现为incremental_sync: type: DatetimeBasedCursor cursor_field: date lookback_window: P{{ config.lookback_window }}D cursor_datetime_formats: [%Y-%m-%d] datetime_format: %Y-%m-%d start_datetime: type: MinMaxDatetime datetime: {{ config.start_date }} datetime_format: %Y-%m-%d end_datetime: type: MinMaxDatetime datetime: {{ config.end_date or today_utc() }} datetime_format: %Y-%m-%d step: P1D cursor_granularity: P1D关键点日期格式全部为%Y-%m-%d如2022-11-11起点start_date配置必填终点config.end_date or today_utc()——配置了end_date用配置值含当天文档描述为 Data is retrieved until that date (included)否则用运行当天 UTC 日期切片步长step: P1D表示按每天切分请求区间即每个 slice 覆盖一天对应 DAILY 粒度回看窗口lookback_window: P{{ config.lookback_window }}D让每次增量都额外向前回看若干天以吸收 Apple 归因数据的延迟更新见配置参数表。4.1 从分区状态到全局游标1.0.0 破坏性变更从源码结构看adgroups_report_daily、keywords_report_daily、ads_report_daily三个流都声明了global_substream_cursor: true。这与 metadata.yaml 中记录的 1.0.0 破坏性变更直接对应该版本将adgroups_report_daily与keywords_report_daily的状态从按分区per-partition状态改为使用全局状态游标global state cursor从而缩短这两个流的读取时间。campaigns_report_daily未在变更影响范围内说明其从更早版本起即使用全局游标。对于从旧版本升级的用户需要清空clear受影响流的历史数据后再同步迁移截止时间为 2025-11-04。这一升级细节是排查升级后状态不兼容问题的重要线索。五、配置参数详解连接器的spec定义在 manifest.yaml 的spec.connection_specification中。必填项为org_id、client_id、start_date、client_secret、timezone、token_refresh_endpoint。可参考 sample_config.json 中的示例结构{ org_id: REPLACEME, client_id: REPLACEME, client_secret: REPLACEME, token_refresh_endpoint: https://apple.oauth.com/token, start_date: 2022-11-11, backoff_factor: 10, lookback_window: 3 }参数类型必填默认值说明org_idinteger是—拥有 Campaign 的组织标识符与 Apple Search Ads 控制台中的账户Org一致随请求头X-AP-Context发送client_idstring是—获取令牌所用的用户标识OAuth client idsecret 类型client_secretstring是—认证用户设置请求的客户端密钥secret 类型start_datestring是—首次同步数据的起始日期格式YYYY-MM-DD正则^[0-9]{4}-[0-9]{2}-[0-9]{2}$end_datestring否当前日期数据检索截至日期含当天格式同start_date未设置则以运行当天 UTC 为终点timezonestring是UTC报告统计时区仅UTC或ORTZ组织时区token_refresh_endpointstring是https://appleid.apple.com/auth/oauth2/token?grant_typeclient_credentialsscopesearchadsorgOAuth 令牌刷新端点需要代理 Apple 令牌请求时可覆盖backoff_factorinteger否5指数退避的延迟增长系数有效值 1–20正则^(20|1[0-9]|[1-9])$lookback_windowinteger否30增量同步回看天数Apple 采用 30 天归因窗口调小可缩短同步耗时代价是可能漏掉延迟归因数据示例值为 7num_workersinteger否2同步并发工作线程数范围 1–20配合concurrency_level默认取num_workers上限 20使用其中start_date/end_date的正则约束、num_workers的 1–20 范围、backoff_factor与lookback_window的取值范围都能在 manifest 的 spec 中直接找到依据。sample_config.json中的token_refresh_endpoint是测试环境端点生产环境应按需使用默认的 Apple 端点。六、认证与错误处理6.1 OAuth client_credentials所有请求都通过OAuthAuthenticator完成认证manifest 的base_requesterauthenticator: type: OAuthAuthenticator client_id: {{ config.client_id }} grant_type: client_credentials client_secret: {{ config.client_secret }} token_refresh_endpoint: {{ config.get(token_refresh_endpoint, https://appleid.apple.com/auth/oauth2/token?grant_typeclient_credentialsscopesearchadsorg) }} refresh_request_body: {}即使用client_credentials授权模式用client_idclient_secret换取访问令牌端点可在token_refresh_endpoint中显式覆盖。6.2 三级重试策略manifest 为每个流都配置了CompositeErrorHandler/DefaultErrorHandler配合HttpResponseFilter401 → 刷新令牌后重试action: REFRESH_TOKEN_THEN_RETRYfailure_type: transient_error错误信息为 Access token is expired.——当 Apple 在 CDK 记录的令牌过期时间之前就返回 401 时连接器会先主动刷新令牌再重试500 / 429 → 直接重试action: RETRY其中 429 是限流、500 是服务端错误指数退避ExponentialBackoffStrategy按config.backoff_factor控制延迟增长报告流max_retries: 10的重试上限高于实体流。此外keywords_report_daily有一个特殊过滤规则当错误消息包含CAMPAIGN DOES NOT CONTAIN KEYWORD时执行IGNORE——即该 Campaign 不含关键词并非异常直接跳过该切片避免因个别空数据分区导致整个同步失败。这些行为已被 unit_tests/test_manifest.py 固化为断言如test_streams_reactively_refresh_oauth_token_on_401校验每个流都具备 401 刷新重试、test_keywords_report_daily_retains_keyword_predicate校验关键词流的 IGNORE 谓词、test_ads_report_daily_no_keyword_error_predicate则防止该谓词被错误地复制到 ads 报告流。七、分页与并发实体流分页采用OffsetIncrement策略通过查询参数offset与limit翻页每页page_size: 1000报告流分页同样的 offset/limit 策略但注入位置是请求体 JSON 的selector.pagination.offset/selector.pagination.limit并发控制manifest 顶层声明ConcurrencyLevel默认并发数为config.get(num_workers, 2)上限 20。num_workers默认 2、范围 1–20。由于 Keywords / Ads 属于两层嵌套的子流先遍历 Campaign 再遍历 AdGroup并发分区处理能显著缩短深层嵌套流的同步时间——test_concurrency_level_configured明确注释了这是为了防止深层嵌套子流在心跳超时内无法完成。八、测试与质量保障连接器的测试配置展示了针对该数据源的实际验证手段unit_tests/test_manifest.py以数据驱动方式校验 manifest 的日期字段变换date必须取自stream_slice.start_time而非stream_slice.start_date、401 刷新、关键词谓词、并发配置与num_workersspec 字段acceptance-test-config.yml声明spec与connection两类验收测试其中连接测试使用无效配置invalid_config.json 包含wrongid、非法日期9999-99-99、非法backoff_factor等验证校验器能正确拒绝full_refresh 验收测试对granularity与metadata字段做了天然不可幂等的豁免说明integration_tests/sample_state.json 与 abnormal_state.json分别给出正常游标date: 2022-11-09与未来游标date: 2999-11-06用于验证增量同步在正常与超前状态下的行为。九、小结同步行为速查数据流同步模式游标主键请求方式campaignsfull_refresh—idGET/campaignsadgroupsfull_refresh—idGET/campaigns/{id}/adgroupskeywordsfull_refresh—idGET/campaigns/{cid}/adgroups/{aid}/targetingkeywordsadsfull_refresh—idGET/campaigns/{cid}/adgroups/{aid}/adscampaigns_report_dailyfull_refresh incrementaldatedate campaignIdPOST/reports/campaignsadgroups_report_dailyfull_refresh incrementaldate全局游标date adGroupIdPOST/reports/campaigns/{id}/adgroupskeywords_report_dailyfull_refresh incrementaldate全局游标date keywordIdPOST/reports/campaigns/{id}/keywordsads_report_dailyfull_refresh incrementaldate全局游标date adIdPOST/reports/campaigns/{id}/ads实际使用时实体类数据建议配合报告流一起同步实体流提供 Campaign / AdGroup / Keyword / Ad 的元数据与 ID 映射报告流按 DAILY 粒度持续增量拉取统计指标start_date决定初始回填范围end_date不设则默认到当天lookback_window用于兜底 Apple 延迟归因的数据修正。深入阅读 bootstrap.md 与 manifest.yaml 可获得与本文完全一致的权威细节。赞分享数据工程数据集成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 Bing Ads Source 连接器深度解析从 OAuth 认证、账户分层流到报告与 Bulk 异步下载的完整实现指南Airbyte Bing Ads Source 连接器深度解析从 OAuth 认证、账户分层流到报告与 Bulk 异步下载的完整实现指南 本篇技术指南以 Ai数据工程数据集成ETL后端大数据Airbyte source-amazon-ads 连接器深度剖析异步报告生成、HTTP 425 冲突与增量同步的独特行为Airbyte source amazon ads 连接器深度剖析异步报告生成、HTTP 425 冲突与增量同步的独特行为 本篇技术指南围绕 Airbyte数据工程数据集成ETL后端大数据Airbyte source-bing-ads 连接器独特行为深度解析谓词去重过滤器与 Bulk 报告 gzip 解码回退机制Airbyte source bing ads 连接器独特行为深度解析谓词去重过滤器与 Bulk 报告 gzip 解码回退机制 本指南以 source bin数据工程数据集成ETL后端大数据创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考