Mage AI 数据集成：Twitter Ads 源连接器配置指南与同步原理详解

数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载本指南聚焦 Mage AImage-ai开源仓库中 Twitter Ads 数据源连接器的完整使用方式。你将掌握该连接器所需的全部配置参数OAuth 1.0a 凭据、账户 ID、报告定义等、如何申请 Twitter Ads API 访问权限以及从源码层面理解其数据同步、增量书签bookmark与异步报表抽取的底层实现。阅读后可直接在 Mage 的数据集成管道中落地 Twitter Ads 数据抽取任务。连接器概览Twitter Ads 源连接器用于从 TwitterX广告平台拉取营销数据支持两大类型的数据实体对象流Streams账户、广告系列、广告组line items、推广推文、卡片、受众等投放管理对象报表流Reports通过reports配置定义的、按实体/维度/粒度聚合的分析报表例如按性别、地区、设备等维度拆分的投放效果数据。连接器基于 Singer 标准singer库实现位于仓库 mage_integrations/mage_integrations/sources/twitter_ads 目录其入口类为TwitterAds(Source)定义于init.py实现了discover发现数据目录、sync执行同步与test_connection校验凭据三个核心生命周期方法。配置参数在 Mage 中配置该源时必须提供以下凭据与参数。完整示例模板可参考 templates/config.json。参数说明示例值start_date增量同步书签bookmark端点的绝对起始时间YYYY-MM-DDTHH:MM:SSZ。2023-01-01T00:00:00Zconsumer_keyOAuth 1.0a 消费者密钥。YOUR_TWITTER_ADS_CONSUMER_KEYconsumer_secretOAuth 1.0a 消费者密钥对应的 Secret。YOUR_TWITTER_ADS_CONSUMER_SECRETaccess_tokenOAuth 1.0a 访问令牌。YOUR_TWITTER_ADS_ACCESS_TOKENaccess_token_secretOAuth 1.0a 访问令牌 Secret。YOUR_TWITTER_ADS_ACCESS_TOKEN_SECRETaccount_ids逗号分隔的 Twitter 广告账户 ID 列表。id1, id2, id3attribution_window归因回看窗口天数用于等待分析报表数据稳定后再抽取默认示例为14。14with_deletedtrue或false是否在结果中保留逻辑删除记录。truecountry_codes逗号分隔的 ISO 两位国家代码用于定向targeting与细分segmentation。US, CA, MX, DEpage_size可选参数自定义分页大小。1000reports报表定义对象数组每个报表包含name、entity、segment、granularity。[{name: campaigns_genders_hourly_report, entity: CAMPAIGN, segment: GENDER, granularity: HOUR}]request_timeoutTwitter Ads 客户端的连接与读取超时时间默认 300 秒。300官方配置文档同样收录于 docs/data-integrations/sources/twitter_ads.mdx。必填项与可选配置从源码看tap_twitter_ads/init.py 中定义了REQUIRED_CONFIG_KEYS以下六个键为启动同步所必需start_dateconsumer_keyconsumer_secretaccess_tokenaccess_token_secretaccount_ids其余参数attribution_window、with_deleted、country_codes、page_size、reports、request_timeout均为可选项但会影响抽取范围与行为attribution_window影响报表回看的绝对起始时间计算。在 streams.py 的get_absolute_start_end_time中若当前时间与上次书签的间隔天数小于归因窗口则回看起点会被强制前移为「当前时间 - attribution_window 天」以保证归因数据已稳定。with_deleted默认true。会被注入到各端点请求参数中见 streams.py注意 Twitter Ads API 只接受小写true/false代码中也是以小写字符串拼接。country_codes用于targeting_events等定向选项端点以及targeting_locations、targeting_network_operators等按国家循环的子类型sub_type端点同时用于报表中LOCATIONS、REGIONS、METROS、POSTAL_CODES分段的国家 ID 解析见 sync.py。page_size用于覆盖各端点请求参数中的count值。校验逻辑见 streams.py空字符串回退到默认值非整数或小于等于 0 的值会抛出The entered page size ({}) is invalid异常。request_timeout最终传递给 SDK 客户端构造函数的timeout选项0、空字符串等非法值会回退到默认 300 秒见 tap_twitter_ads/init.py。reports定义需要抽取的报表流缺省为空列表即不抽取任何报表。获取 Twitter Ads API 访问权限使用本连接器前必须先申请 Twitter Ads API 访问权限。大致流程为在 TwitterX开发者平台创建开发者应用App为应用申请 Twitter Ads API 的访问等级与产品权限在应用管理页面中生成 OAuth 1.0a 凭据Consumer Key / Consumer Secret以及访问令牌Access Token / Access Token Secret将四组凭据与广告账户 ID 填入 Mage 的源配置中。官方入门指引请参考 Twitter 开发者文档中的 Twitter Ads API Getting Started 指南当前仓库 README 与 docs/data-integrations/sources/twitter_ads.mdx 均给出该链接指向。凭据校验机制配置完成后Mage 会调用test_connection见init.py对连接做预检。其内部实现tap_twitter_ads/init.py包含两个步骤调用get_resource(accounts, client, accounts)验证令牌是否有效逐个校验account_ids中每个账户是否可访问无效账户会被收集并统一抛出Invalid Twitter Ads accounts provided during the configuration: [...]异常。一个完整的配置文件示例以下 JSON 综合了 templates/config.json 与文档参数说明可作为在 Mage 中配置该源的直接参考{ start_date: 2019-01-01T00:00:00Z, consumer_key: YOUR_TWITTER_ADS_CONSUMER_KEY, consumer_secret: YOUR_TWITTER_ADS_CONSUMER_SECRET, access_token: YOUR_TWITTER_ADS_ACCESS_TOKEN, access_token_secret: YOUR_TWITTER_ADS_ACCESS_TOKEN_SECRET, account_ids: id1, id2, id3, attribution_window: 14, with_deleted: true, country_codes: US, CA, MX, DE, page_size: 1000, reports: [ { name: campaigns_genders_hourly_report, entity: CAMPAIGN, segment: GENDER, granularity: HOUR }, { name: line_items_regions_daily_report, entity: LINE_ITEM, segment: REGIONS, granularity: DAY } ], request_timeout: 300 }注意模板中reports对象的entity键在文档示例中写作enitity原文档笔误配置时应使用规范拼写entity否则无法通过 schema.py 中的实体校验会抛出INVALID ENTITY错误。支持的流Streams连接器在目录发现discover阶段会枚举 streams.py 中STREAMS字典注册的全部流共 35 个端点含父流与子流。按其数据特性可分为三类1. 投放管理实体流增量同步以下流使用INCREMENTAL增量复制方式复制键replication key为updated_at主键为idaccounts、account_media、campaigns、funding_instruments、line_items、media_creatives、preroll_call_to_actions、promoted_accounts、promoted_tweets、promotable_users、scheduled_promoted_tweets、tailored_audiences、tracking_tags、cards、cards_poll、cards_image_conversation、cards_video_conversation。请求参数统一包含sort_by: [updated_at-desc]、with_deleted: {with_deleted}、count与cursor即按更新时间倒序拉取便于增量书签处理。其中cards端点的count上限被注释为 200API 对超过 200 的 page size 会报错其余多数端点默认 1000见 streams.py。2. 定向选项流全量同步以下流使用FULL_TABLE全量复制方式数据量相对稳定用于同步 Twitter 定向投放的选项字典advertiser_business_categories、content_categories、iab_categories、targeting_app_store_categories、targeting_conversations、targeting_devices、targeting_events、targeting_interests、targeting_languages、targeting_locations、targeting_network_operators、targeting_platforms、targeting_platform_versions、targeting_tv_markets、targeting_tv_shows。其中targeting_locations与targeting_network_operators会依据country_codes配置逐个国家循环请求sub_types [{country_code_list}]targeting_tv_shows是targeting_tv_markets的子流按locale关联parent_ids_limit 1。3. 特殊流与父子流关系tweets增量流复制键为created_at时间格式%a %b %d %H:%M:%S %z %Y包含PUBLISHED与SCHEDULED两个子类型两者分别维护独立书签见 streams.py。line_items→targeting_criteriatargeting_criteria是line_items的子流按line_item_ids批量查询parent_ids_limit 200主键为[line_item_id, id]复合键。targeting_tv_markets→targeting_tv_shows如上所述按 locale 关联。schema.py中get_schemas会为每个流加载对应的 JSON Schema位于 tap_twitter_ads/schemas共 40 余个流与共享定义文件并将复制键标记为automatic自动包含。若仅选中子流而未选中父流同步逻辑也会自动把父流加入同步队列保证子流数据可正确关联见 sync.py。配置报表Reportsreports参数是抽取聚合分析数据的关键每个报表对象包含四个字段字段说明可选值以源码为准name报表在目录中的唯一名称例如campaigns_genders_hourly_report。自定义字符串entity报表统计的实体类型。ACCOUNT、CAMPAIGN、FUNDING_INSTRUMENT、LINE_ITEM、MEDIA_CREATIVE、ORGANIC_TWEET、PROMOTED_TWEET、PROMOTED_ACCOUNTsegment细分维度NO_SEGMENT表示不细分。NO_SEGMENT、AGE、GENDER、DEVICES、LOCATIONS、REGIONS、METROS、POSTAL_CODES、PLATFORMS、PLATFORM_VERSIONS、LANGUAGES、INTERESTS、KEYWORDS、CONVERSATIONS、CONVERSION_TAGS、AUDIENCES、EVENTS、TV_SHOWS等完整列表见 schema.pygranularity时间粒度。HOUR、DAY、TOTAL报表定义校验规则schema.py 在发现阶段会对报表组合做严格校验任一规则不满足都会直接抛出运行时错误entity、segment、granularity必须属于上述枚举MEDIA_CREATIVE与ORGANIC_TWEET不允许任何细分segment 必须为NO_SEGMENTCONVERSION_TAGS细分仅允许ACCOUNT、CAMPAIGN、LINE_ITEM、PROMOTED_TWEET实体使用LANGUAGES细分不允许ACCOUNT、FUNDING_INSTRUMENT、MEDIA_CREATIVE实体使用。此外报表流的 Schema 会依据组合动态选择CONVERSION_TAGS只允许WEB_CONVERSION指标组加载report_web_conversion.jsonACCOUNT、FUNDING_INSTRUMENT、ORGANIC_TWEET只能使用各自受限的指标组见 schema.py。NO_SEGMENT报表会移除dimensions中的细分字段非NO_SEGMENT/PLATFORMS/CONVERSION_TAGS组合会移除web_conversion字段。同步流程与增量书签原理实体流的同步链路sync主流程sync.py的组织方式为解析配置将account_ids、country_codes按逗号拆分并去除空格从目录中提取用户选中的流并区分父流与子流子流未选中时自动补充其父流账户外层循环对每个账户依次执行「父流同步 → 报表所需国家/平台 targeting ID 解析 → 报表流同步」通过update_currently_syncing维护状态中的currently_syncing字段若同步中途中断可从上次流位置恢复。每个实体流的实际请求与翻页由 streams.py 的sync_endpoint完成将{account_id}、{with_deleted}、{parent_ids}、{start_date}、{country_codes}、{sub_type}等占位符替换为实际值后发起请求请求返回 Cursor 对象记录按复制键倒序排列首条记录即为当前批次的最大书签值get_maximum_bookmark当某条记录的复制键小于上次保存的书签时停止拉取增量截断每条记录经transform_record处理、追加account_id字段再经 SingerTransformer按 Schema 校验/脱敏后写出子流按父流 ID 分批chunk关联查询例如targeting_criteria每 200 个line_item_ids一批。按账户维度保存书签书签状态以account_id为键分层保存state[bookmarks][stream][account_id]见 streams.py。tweets流进一步按PUBLISHED/SCHEDULED子类型分别记录书签。这意味着多个广告账户可以在同一次同步中各自独立维护增量进度。报表流的异步抽取机制报表数据量较大Twitter Ads API 采用异步任务方式提供。Reports.sync_reportstreams.py的处理流程为日期窗口循环从书签时间到当前时间按窗口推进——有细分segment时窗口为 42 天无细分时为 85 天低于 API 的 45/90 天上限避免小时/日期取整问题对每个时间窗口按ACCOUNT直接用账户 ID、ORGANIC_TWEET从 tweets 流获取实体 ID或其余实体调用active_entities端点解析出活跃实体 ID 集合异步任务提交实体 ID 每 20 个一组向stats/jobs/accounts/{account_id}POST 异步任务携带entity、entity_ids、metric_groups、placementALL_ON_TWITTER/PUBLISHER_NETWORK两种投放位置循环、granularity、start_time、end_time以及可选的segmentation_type、country、platform参数轮询任务状态每 15 秒查询一次任务状态最多 20 次SUCCESS的任务将返回结果下载 URL见get_async_results_urlsstreams.py下载与写出从 URL 下载数据经transform_report处理后写出报表流以记录中的end_time为复制键逐条比较更新最大书签值。指标组Metric Groups的选择不同实体允许的指标组组合在get_entity_metric_groupsstreams.py中定义CAMPAIGN、LINE_ITEM、PROMOTED_TWEET、PROMOTED_ACCOUNT、MEDIA_CREATIVE支持全部指标组ENGAGEMENT、BILLING、VIDEO、MEDIA、WEB_CONVERSION、MOBILE_CONVERSION、LIFE_TIME_VALUE_MOBILE_CONVERSIONACCOUNT仅ENGAGEMENTFUNDING_INSTRUMENTENGAGEMENT、BILLINGORGANIC_TWEETENGAGEMENT、VIDEOCONVERSION_TAGS细分时仅WEB_CONVERSION。客户端行为与错误处理连接器底层使用 Twitter 官方 Python Ads SDK 的Client。构造参数见 tap_twitter_ads/init.py包含handle_rate_limit: True自动处理 429 限流retry_max: 10最多重试 10 次retry_delay: 60000每次重试等待 1 分钟毫秒retry_on_status: [400, 420, 500, 502, 503, 504]这些状态码触发重试retry_on_timeouts: True超时也触发重试timeout取自request_timeout配置默认 300 秒。此外streams.py 使用backoff对 SDK 的Request.perform做了装饰遇到ConnectionError时按恒定间隔60 秒最多重试 5 次。HTTP 错误会按状态码映射为语义化异常见 client.py状态码异常类型含义400TwitterAdsBadRequestError请求缺失参数或参数错误401TwitterAdsUnauthorizedError访问未授权403TwitterAdsForbiddenError用户无权访问该资源404TwitterAdsNotFoundError指定的资源不存在405TwitterAdsMethodNotFoundErrorURL 不支持该 HTTP 方法408TwitterAdsRequestCancelledError请求被取消429TwitterAdsClient429Error超过 API 限流请稍后重试500TwitterAdsInternalServerError服务端内部错误503TwitterAdsServiceUnavailableError服务不可用在 Mage 中集成与使用在 Mage 项目中创建「数据集成」类型管道选择Twitter Ads作为 Source按上文参数表填写连接配置推荐直接从templates/config.json复制基础结构在流选择界面勾选需要抽取的实体流与报表流运行连接测试test_connection验证凭据与账户 ID 是否有效保存配置并运行管道Mage 会自动执行目录发现、Schema 写出与增量同步为管道配置调度触发器即可按计划持续将 Twitter Ads 数据同步至目标数据库或数据仓库。若需要更完整地了解 Mage 数据集成管道的通用配置方式包括目标端、I/O 配置与连接管理等可参考仓库 docs/data-integrations/configuration.mdx 与 docs/data-integrations/overview.mdx。小结Twitter Ads 源连接器是 Mage 数据集成生态中面向广告营销数据的标准入口。通过本文介绍的配置参数、报表定义规则与源码级同步原理你可以准确完成连接器配置理解书签与异步报表机制并在生产管道中稳定、增量地抽取 Twitter 广告数据。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐从混用尺寸到统一 A4PDF补丁丁页面大小调整的 3 个实操任务从混用尺寸到统一 A4PDF补丁丁页面大小调整的 3 个实操任务 扫描文档一半 A4 一半 A5合并后的文件页面尺寸各不相等打印排队时最头疼。PDF补丁丁数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage AI 数据集成Delta Lake (Azure) 目标连接器配置与实现原理指南Mage AI 数据集成Delta Lake Azure 目标连接器配置与实现原理指南 导读 本文围绕 Mage AI 开源仓库中 Delta Lake Az数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理Mage 数据集成中接入 Outreach 数据源OAuth 认证配置、参数详解与增量同步原理 Outreach 是销售参与Sales Engagement数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇终极游戏画质升级指南如何用OptiScaler免费解锁显卡超采样技术下一篇ODM教育版为学术机构定制的开源无人机数据处理方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考