企业官网建设流程全解析

• 数据工程
• 数据编排
• ETL
• 任务调度
• 批处理
• 流处理
• 数据集成
• 后端

本指南聚焦 Mage AI（mage-ai）开源仓库中 Twitter Ads 数据源连接器的完整使用方式。你将掌握该连接器所需的全部配置参数（OAuth 1.0a 凭据、账户 ID、报告定义等）、如何申请 Twitter Ads API 访问权限，以及从源码层面理解其数据同步、增量书签（bookmark）与异步报表抽取的底层实现。阅读后可直接在 Mage 的数据集成管道中落地 Twitter Ads 数据抽取任务。

连接器概览

Twitter Ads 源连接器用于从 Twitter（X）广告平台拉取营销数据，支持两大类型的数据：

• 实体对象流（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。

官方配置文档同样收录于 docs/data-integrations/sources/twitter_ads.mdx。

必填项与可选配置

从源码看，tap_twitter_ads/init.py 中定义了REQUIRED_CONFIG_KEYS，以下六个键为启动同步所必需：

• start_date
• consumer_key
• consumer_secret
• access_token
• access_token_secret
• account_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 访问权限。大致流程为：

1. 在 Twitter（X）开发者平台创建开发者应用（App）；
2. 为应用申请 Twitter Ads API 的访问等级与产品权限；
3. 在应用管理页面中生成 OAuth 1.0a 凭据（Consumer Key / Consumer Secret）以及访问令牌（Access Token / Access Token Secret）；
4. 将四组凭据与广告账户 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）包含两个步骤：

1. 调用get_resource('accounts', client, 'accounts')验证令牌是否有效；
2. 逐个校验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，主键为id：

accounts、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上限被注释为 200（API 对超过 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_criteria：targeting_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）。

配置报表（Reports）

reports参数是抽取聚合分析数据的关键，每个报表对象包含四个字段：

报表定义校验规则

schema.py 在发现阶段会对报表组合做严格校验，任一规则不满足都会直接抛出运行时错误：

• entity、segment、granularity必须属于上述枚举；
• MEDIA_CREATIVE与ORGANIC_TWEET不允许任何细分（segment 必须为NO_SEGMENT）；
• CONVERSION_TAGS细分仅允许ACCOUNT、CAMPAIGN、LINE_ITEM、PROMOTED_TWEET实体使用；
• LANGUAGES细分不允许ACCOUNT、FUNDING_INSTRUMENT、MEDIA_CREATIVE实体使用。

此外，报表流的 Schema 会依据组合动态选择：CONVERSION_TAGS只允许WEB_CONVERSION指标组（加载report_web_conversion.json）；ACCOUNT、FUNDING_INSTRUMENT、ORGANIC_TWEET只能使用各自受限的指标组（见 schema.py）。NO_SEGMENT报表会移除dimensions中的细分字段，非NO_SEGMENT/PLATFORMS/CONVERSION_TAGS组合会移除web_conversion字段。

同步流程与增量书签原理

实体流的同步链路

sync主流程（sync.py）的组织方式为：

1. 解析配置：将account_ids、country_codes按逗号拆分并去除空格；
2. 从目录中提取用户选中的流，并区分父流与子流（子流未选中时自动补充其父流）；
3. 账户外层循环：对每个账户依次执行「父流同步 → 报表所需国家/平台 targeting ID 解析 → 报表流同步」；
4. 通过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_report（streams.py）的处理流程为：

1. 日期窗口循环：从书签时间到当前时间按窗口推进——有细分（segment）时窗口为 42 天，无细分时为 85 天（低于 API 的 45/90 天上限，避免小时/日期取整问题）；
2. 对每个时间窗口，按ACCOUNT（直接用账户 ID）、ORGANIC_TWEET（从 tweets 流获取实体 ID）或其余实体（调用active_entities端点）解析出活跃实体 ID 集合；
3. 异步任务提交：实体 ID 每 20 个一组，向stats/jobs/accounts/{account_id}POST 异步任务，携带entity、entity_ids、metric_groups、placement（ALL_ON_TWITTER/PUBLISHER_NETWORK两种投放位置循环）、granularity、start_time、end_time以及可选的segmentation_type、country、platform参数；
4. 轮询任务状态：每 15 秒查询一次任务状态（最多 20 次），SUCCESS的任务将返回结果下载 URL（见get_async_results_urls，streams.py）；
5. 下载与写出：从 URL 下载数据，经transform_report处理后写出；报表流以记录中的end_time为复制键，逐条比较更新最大书签值。

指标组（Metric Groups）的选择

不同实体允许的指标组组合在get_entity_metric_groups（streams.py）中定义：

• CAMPAIGN、LINE_ITEM、PROMOTED_TWEET、PROMOTED_ACCOUNT、MEDIA_CREATIVE：支持全部指标组（ENGAGEMENT、BILLING、VIDEO、MEDIA、WEB_CONVERSION、MOBILE_CONVERSION、LIFE_TIME_VALUE_MOBILE_CONVERSION）；
• ACCOUNT：仅ENGAGEMENT；
• FUNDING_INSTRUMENT：ENGAGEMENT、BILLING；
• ORGANIC_TWEET：ENGAGEMENT、VIDEO；
• CONVERSION_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）：

在 Mage 中集成与使用

1. 在 Mage 项目中创建「数据集成」类型管道，选择Twitter Ads作为 Source；
2. 按上文参数表填写连接配置（推荐直接从templates/config.json复制基础结构）；
3. 在流选择界面勾选需要抽取的实体流与报表流；
4. 运行连接测试（test_connection）验证凭据与账户 ID 是否有效；
5. 保存配置并运行管道，Mage 会自动执行目录发现、Schema 写出与增量同步；
6. 为管道配置调度触发器，即可按计划持续将 Twitter Ads 数据同步至目标数据库或数据仓库。

若需要更完整地了解 Mage 数据集成管道的通用配置方式（包括目标端、I/O 配置与连接管理等），可参考仓库 docs/data-integrations/configuration.mdx 与 docs/data-integrations/overview.mdx。

小结

Twitter Ads 源连接器是 Mage 数据集成生态中面向广告营销数据的标准入口。通过本文介绍的配置参数、报表定义规则与源码级同步原理，你可以准确完成连接器配置，理解书签与异步报表机制，并在生产管道中稳定、增量地抽取 Twitter 广告数据。

• 数据工程
• 数据编排
• ETL
• 任务调度
• 批处理
• 流处理
• 数据集成
• 后端