数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载LinkedIn Ads领英广告是营销数据集成场景中的高频数据源。本文以 mage-ai 开源仓库中的 linkedin_ads 数据源模块 为主线完整讲解该数据源的配置项、两种 OAuth 认证方式、8 个可用数据流stream以及底层增量同步、分页与错误重试机制。读完本文你将能够在 Mage 项目中独立配置并跑通 LinkedIn Ads → 数据仓库/数湖的营销数据同步管道并理解配置项背后的源码实现。一、数据源概述linkedin_ads是 Mage 数据集成框架mage_integrations中的一个 Singer 风格数据源tap对应的主类为LinkedinAds。它封装了 LinkedIn Marketing Developer Platform 的 Ads APIhttps://api.linkedin.com/v2对外暴露三个标准能力discover(...)调用 discover.py 读取本地 schema 并生成可选的 catalog 流sync(...)调用 sync.py 执行真实的数据拉取test_connection()调用client.check_accounts(config)校验配置中提供的广告账号是否有效。从源码结构看LinkedinAds直接继承mage_integrations.sources.base.Source因此它可以被 Mage 的数据集成管道data integration pipeline直接使用并复用框架提供的 catalog、schema 与状态管理能力。二、配置参数详解按照 README.md 的说明配置该数据源时必须填写以下凭据字段。完整的模板文件位于 templates/config.json可直接作为配置起点{ client_id: , client_secret: , refresh_token: , accounts: , request_timeout: 300, start_date: 2023-01-01T00:00:00Z, user_agent: }2.1 基础字段Key说明示例值是否必填accounts需要同步的 LinkedIn 广告账号 ID 列表逗号分隔仅当你想同步accounts或account_users流时需要id1, id2, id3同步accounts/account_users时必填access_token长期访问令牌def789...✅request_timeout单次 API 请求超时时间秒300选填start_date增量同步的起始时间ISO 8601 格式2023-01-01T00:00:00Z✅user_agent请求头中的 User-Agent建议填联系邮箱your_emailyour_domain.com✅其中request_timeout与源码中的常量REQUEST_TIMEOUT 300client.py保持一致。在 LinkedinClient 构造函数 中可以看到其解析逻辑当request_timeout传入的值为非 0 的数值时转换为float作为真实超时时间当值为0、0或空字符串时回退到默认的 300 秒。注意accounts字段在同步时会被按逗号拆分并去除空格config[accounts].replace( , ).split(,)见 sync.py因此示例值中的空格是可容忍的。2.2 替代认证字段OAuth 三件套除了直接提供access_tokenREADME 还给出了另一种更推荐的长效认证方式——提供 OAuth 应用的客户端凭据与刷新令牌Key说明示例值是否必填client_idLinkedIn 应用的客户端 IDabc123...✅client_secretLinkedIn 应用的客户端密钥xyz456...✅refresh_token刷新令牌用于自动续期 access_tokendef789...✅这两种认证方式在 LinkedinClient 中统一处理当未提供refresh_token时视为“旧连接”直接信任用户传入的access_token当提供了refresh_token时客户端会在进入上下文__enter__时自动调用fetch_and_set_access_token()判断令牌是否过期并刷新。2.3 如何获取 access_token官方流程按 README 的步骤指引获取access_token的完整流程为登录 LinkedIn 开发者平台创建一个 LinkedIn 应用在应用中启用Marketing Developer Platform产品该产品需要单独申请填写接入申请表并提交等待数个工作日的审核批准审批通过后使用开发者平台的 OAuth 工具按指引生成 access token。提示由于 Marketing Developer Platform 属于受限产品审核通常需要数天。在等待期内可以先在 Mage 中配置client_id、client_secret、refresh_token三件套并接入同步逻辑待令牌可用后再跑通全流程。三、令牌生命周期管理从源码看自动续期linkedin_ads数据源的核心健壮性设计集中在 client.py 的令牌管理逻辑中这也是它区别于“一次性 access_token”方案的关键。3.1 令牌刷新流程客户端维护了三个 OAuth 相关端点常量BASE_URL https://api.linkedin.com/v2Ads API 根地址LINKEDIN_TOKEN_URI https://www.linkedin.com/oauth/v2/accessToken令牌刷新端点INTROSPECTION_URI https://www.linkedin.com/oauth/v2/introspectToken令牌校验端点。fetch_and_set_access_token()client.py的执行逻辑为若未配置refresh_token直接返回视为已提供有效 access_token 的旧连接若已配置 access_token则调用get_token_expires()调用 introspect 接口获取令牌过期时间若expires_at晚于当前时间则日志记录“令牌仍有效”并复用现有令牌否则调用refresh_access_token()向accessToken端点提交grant_typerefresh_token换取新的 access_token并按返回的expires_in秒推算新的过期时间。令牌刷新与校验方法均带有backoff.on_exception指数退避重试max_tries5, factor2针对Server5xxError与LinkedInUnauthorizedError自动重试。3.2 请求统一入口与重试策略所有 API 请求统一走request()方法client.py它会在每次请求前再次调用fetch_and_set_access_token()确保令牌始终有效随后自动注入Authorization: Bearer token与Accept: application/json请求头POST 请求额外注入Content-Type。针对不同失败类型重试策略分为两套对 5xx 服务端错误、连接错误与 429 限流使用max_time60010 分钟配合full_jitter的全抖动退避——源码注释指出这是为了应对 LinkedIn 报告 API 的“每 5 分钟 4500 万指标值”的数据节流限制对requests.exceptions.Timeout超时错误使用max_tries5, factor2的退避重试。3.3 错误码与语义化异常client.py 维护了一张ERROR_CODE_EXCEPTION_MAPPING将常见 HTTP 状态码映射为语义明确的异常类型HTTP 状态码异常类型语义400LinkedInBadRequestError请求缺少参数或参数错误401LinkedInUnauthorizedError认证凭据无效403LinkedInForbiddenError用户无访问该资源权限404LinkedInNotFoundError账号无效或无权访问该广告账号405LinkedInMethodNotAllowedErrorHTTP 方法不支持411LinkedInLengthRequiredError缺少 Content-Length 头429LinkedInRateLimitExceeededError触发 API 限流500LinkedInInternalServiceErrorLinkedIn 服务端错误504LinkedInGatewayTimeoutError网关超时值得注意的细节当响应码为 401 且错误描述包含Expired access token时日志会输出明确提示——令牌已按 LinkedIn 安全策略过期需要重新认证连接以生成新令牌client.py。同时 404 响应会被特殊处理为自定义提示信息避免直接暴露 Not Found 这种无意义的原始文案。四、支持的 8 个数据流Streams数据源共定义了 8 个流其主键、复制方法与复制键在 schema.py 中统一定义对应的 JSON Schema 存放在 tap_linkedin_ads/schemas/ 目录下每个流一个.json文件。流名称主键key_properties复制方法复制键replication_keysaccountsidINCREMENTALlast_modified_timevideo_adscontent_referenceINCREMENTALlast_modified_timeaccount_usersaccount_id,user_person_idINCREMENTALlast_modified_timecampaign_groupsidINCREMENTALlast_modified_timecampaignsidINCREMENTALlast_modified_timecreativesidINCREMENTALlast_modified_timead_analytics_by_campaigncampaign_id,start_atINCREMENTALend_atad_analytics_by_creativecreative_id,start_atINCREMENTALend_at8 个流全部采用INCREMENTAL增量复制对象类流accounts、campaigns、creatives 等以last_modified_time为增量键两个广告分析流ad_analytics_by_campaign/ad_analytics_by_creative以end_at为增量键。测试基类 tests/base.py 中对上述元数据做了完整的断言校验可作为理解各流语义的权威参考。4.1 流之间的父子关系从 sync.py 的endpoints配置可以看到流之间的依赖层级accounts对应 API 路径adAccountsV2下挂video_ads子流路径adDirectSponsoredContents按账号过滤campaigns路径adCampaignsV2下挂 3 个子流ad_analytics_by_campaignadAnalyticsV2pivotCAMPAIGN按日聚合creativesadCreativesV2按 campaign 搜索ad_analytics_by_creativeadAnalyticsV2pivotCREATIVE按日聚合。同步时父流的每条记录会作为子流请求的过滤条件例如 campaigns 的子流会以search.campaign.values[0]urn:li:sponsoredCampaign:{id}构造查询参数video_ads子流则要求父记录存在reference_organization_id否则会跳过并输出 warningsync.py。4.2 accounts 字段的过滤注入同步时config[accounts]会被注入到不同流的不同查询参数中具体由各流的account_filter类型决定sync.pysearch_id_values_paramaccountssearch.id.values[i]传入整数账号 IDsearch_account_values_paramcampaign_groups、campaignssearch.account.values[i]传入urn:li:sponsoredAccount:{id}accounts_paramaccount_users、两个 analytics 流accounts[i]传入urn:li:sponsoredAccount:{id}。这也解释了 README 中“accounts仅对同步accounts或account_users流为必填”的表述——其余流虽然也可以指定账号过滤但缺少该字段时依然可以按全账号范围同步。五、增量同步与状态管理机制5.1 bookmark 读写get_bookmark / write_bookmark 实现了 Singer 标准的 bookmark 状态管理每个流以第一个复制键作为 bookmark 字段写入state[bookmarks]从而支持断点续传。同步开始时以start_date作为默认 bookmark结束后将本批次的最大值写回。5.2 广告分析流的滑动时间窗口ad_analytics_by_campaign与ad_analytics_by_creative是结构最复杂的两个流其同步逻辑独立实现在sync_ad_analytics()sync.py回看窗口LOOKBACK同步起始时间会向前回退LOOKBACK_WINDOW 7天同步代码中 delta7以覆盖广告数据延迟落库的情况时间窗口步长默认DATE_WINDOW_SIZE 30天通过shift_sync_window()逐窗口推进窗口终点不超过今天字段分块请求LinkedIn API 单请求最多返回 20 个字段源码以MAX_CHUNK_LENGTH 17为上限对字段分块并强制在每个分块中附加dateRange、pivot、pivotValue三个字段多响应合并merge_responses()以(pivotValue, dateRange.start)为复合主键将多次分块请求的响应合并为一条完整记录再交给process_records()写入。在 sync.py 中还可以看到FIELDS_AVAILABLE_FOR_AD_ANALYTICS_V2集合它列出了该版本支持请求的全部指标字段clicks、impressions、costInUsd、videoViews、viralShares 等 60 项。同步时只会请求 catalog 中被选中且属于该集合的字段避免请求无效字段。5.3 分页机制对象类流采用start/count游标式分页sync.py默认PAGE_SIZE 100通过响应中paging.links里rel next的href自动翻页广告分析流则通过sync_analytics_endpoint()生成器逐页产出数据。page_size也支持通过配置覆盖config.get(page_size)会改写全局PAGE_SIZE。5.4 数据清洗与字段转换所有原始响应会先经过 transform.py 的transform_json()处理主要转换规则包括驼峰转蛇形convert()将 API 返回的camelCase键名统一转为snake_caseURN 转 IDtransform_urn()将urn:li:sponsoredCampaign:123形式的 URN 解析出整数 ID 字段如campaign_id审计字段上提transform_audit_fields()将嵌套的change_audit_stamps.last_modified.time上提为顶层last_modified_time这正是增量键的来源分析流增强transform_analytics()从嵌套的date_range中生成start_at/end_at并将字符串型金额cost_in_usd等转为 Decimal对象流定制transform_campaigns()扁平化 targeting 结构、transform_creatives()抽象 variables 结构、transform_accounts()转换total_budget金额。六、连接测试与错误排查6.1 账号校验逻辑LinkedinAds.test_connection()最终调用client.check_accounts(config)client.py。其行为是将accounts按逗号拆分后对每个账号调用adAccountUsersV2?qaccountscount1start0accountsurn:li:sponsoredAccount:{id}进行探测返回 400 表示账号 ID 不是合法数字格式返回 404 表示账号是合法数字但并非有效的 LinkedIn 广告账号上述两种情况都会被收集到invalid_account列表最终抛出Invalid Linked Ads accounts provided during the configuration: [...]异常其他非 200 响应走统一错误映射处理。因此如果你在 Mage 界面上配置数据源后连接测试失败优先检查accounts字段中的账号 ID 是否准确、以及该账号是否已授权给当前应用。6.2 测试环境与用例该数据源附带了完整的测试套件tests/unittests/单元测试覆盖账号号码解析test_account_number.py、campaign group 4xx 处理、客户端行为、异常处理、令牌获取、超时重试等tests/ 根目录基于 tap-tester 的集成测试test_all_fields.py、test_discovery.py、test_pagination.py、test_start_date.py、test_sync_canary.py等。集成测试通过环境变量注入凭据见 tests/base.pyTAP_LINKEDIN_ADS_ACCOUNTS广告账号列表TAP_LINKEDIN_ADS_CLIENT_ID/TAP_LINKEDIN_ADS_CLIENT_SECRET应用凭据TAP_LINKEDIN_ADS_REFRESH_TOKEN/TAP_LINKEDIN_ADS_ACCESS_TOKEN令牌。这些测试同时验证了discover阶段返回的流集合必须精确等于上述 8 个流以及各流的 replication 元数据与expected_metadata()完全一致。七、在 Mage 中接入 LinkedIn Ads 数据源在 Mage 项目中接入该数据源的典型路径是创建数据集成管道Data Integration Pipeline选择 LinkedIn Ads 作为源Source填写连接配置按第二节的字段表填写accounts、start_date、user_agent并选择一种认证方式access_token或client_idclient_secretrefresh_token测试连接Mage 会调用test_connection()校验账号有效性选择数据流与字段在 discover 结果中选择需要同步的流建议至少包含campaigns与ad_analytics_by_campaign以覆盖投放与效果两类数据并设置目标表配置调度为管道配置周期触发后续每次运行都会基于 bookmark 只拉取增量数据。由于所有流均为 INCREMENTAL 复制首次同步会从start_date开始全量拉取之后的调度运行仅同步新增与变更数据配合广告分析流的 7 天回看窗口可有效规避广告数据延迟导致的数据空洞。八、小结linkedin_ads数据源是一个完整、健壮的营销数据接入实现配置层面支持长期令牌自动续期运行时内置限流/超时/服务端错误的多层指数退避增量层面通过 bookmark 与滑动时间窗口实现高效同步数据层面完成 URN 转 ID、驼峰转蛇形、金额与时间字段的类型化清洗。无论是作为 Mage 数据集成管道的源还是作为独立 Singer tap 使用本文涉及的配置项与源码机制都能帮助你快速定位问题、稳定产出数据。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐深入解析 Airbyte LinkedIn Pages 声明式连接器manifest 配置、OAuth 认证与增量同步实战深入解析 Airbyte LinkedIn Pages 声明式连接器manifest 配置、OAuth 认证与增量同步实战 LinkedIn Pages 连接数据工程数据集成ETL后端大数据Mage 中配置 Pipedrive 数据源API 认证、可用 Stream 与增量同步原理详解Mage 中配置 Pipedrive 数据源API 认证、可用 Stream 与增量同步原理详解 本文基于 mage ai 开源仓库中的 Pipedrive数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理 Outreach 是销售参与Sales Engagement数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇15DaysofAnimationsinSwift进度动画组件自定义进度条实现原理下一篇Tonic高效简易的C音频合成库创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考