数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载导读本文以 Mage AI 开源仓库中的 Amplitude 连接器文档 为主体结合该连接器的源码实现、Amplitude 数据源集成与事件 JSON Schema系统讲解如何在 Mage AI 中通过 Amplitude Export API 拉取用户行为事件数据。读完本文你将掌握 Amplitude 连接器的核心用法、日期范围语义、分片与解压处理机制以及在 Mage 数据集成管道中配置、运行和增量同步事件的完整方法。快速开始一个最小的拉取示例连接器文档给出的核心用法非常简洁用项目的api_key与secret_key构造连接器实例再调用load()方法即可拉取指定时间范围内的原始事件数据。以下是原文档中的完整示例from connections.amplitude import Amplitude from datetime import datetime, timedelta import json source Amplitude(api_key, secret_key) results source.load(datetime.now() - timedelta(days1)) for r in results[:5]: print(json.dumps(r, indent2))这段代码完成了三件事实例化连接器Amplitude(api_key, secret_key)以 Amplitude 项目的唯一 API Key 与 Secret Key 作为认证凭据内部会调用super().__init__()初始化基类Connection的日志系统见 connections/base.py。指定起始时间load(datetime.now() - timedelta(days1))表示拉取「从昨天到现在」的全部事件。load()的完整签名是load(start_date, end_dateNone, offsetNone, sampleFalse)起始日期必填其余均有默认行为详见下文「日期范围语义」一节。遍历并格式化输出load()返回一个由字典组成的列表每一条对应一条 Amplitude 事件可用json.dumps以缩进格式打印前 5 条做快速巡检。需要说明的是源码中该模块位于mage_integrations/mage_integrations/connections/amplitude/__init__.py在 Mage 的源码目录结构下实际应写作from mage_integrations.connections.amplitude import Amplitude原文档的connections.amplitude是相对其所在包位置的简写。配置项api_key、secret_key 与 hostAmplitude 数据源在 Mage 中的配置参数见 sources/amplitude 文档 与 docs/data-integrations/sources/amplitude.mdx对应连接器构造函数的三个入参参数必填说明示例值api_key是Amplitude 项目唯一的 API Key73bb...secret_key是Amplitude 项目唯一的 Secret KeyABC1...host否Amplitude 服务地址。标准服务为https://amplitude.com欧盟EU Residency服务为https://analytics.eu.amplitude.com默认值为https://amplitude.comhttps://analytics.eu.amplitude.com从源码看connections/amplitude/init.py构造函数定义如下class Amplitude(Connection): def __init__( self, api_key: str, secret_key: str, host: str HOST ): super().__init__() self.api_key api_key self.secret_key secret_key self.host host or HOST其中模块级常量HOST https://amplitude.com为默认主机传入自定义host时用于支持欧盟数据驻留EU Data Residency场景。连接器对 Amplitude 的认证方式采用 HTTP Basic Authrequests.get(url, auth(self.api_key, self.secret_key))即把api_key与secret_key直接作为 Basic Auth 的用户名与密码传给 Export API。在数据源模板中对应配置骨架为templates/config.json{ api_key: , secret_key: }api_key与secret_key都可以在 Amplitude 项目后台的账号凭据页面找到连接器本身只负责按凭据发起请求凭据的创建与权限管理属于 Amplitude 项目侧的能力。日期范围语义start_date、end_date、offset 与 sampleAmplitude Export API 要求按时间范围导出数据因此load()对日期参数的归一化逻辑是连接器最核心的行为之一。它由 connections/amplitude/utils.py 中的build_date_range()统一处理def build_date_range( end_dateNone, offsetNone, sampleFalse, start_dateNone, ): now datetime.today() today datetime(now.year, now.month, now.day, 0) if start_date: if offset is not None: start_date start_date timedelta(hoursoffset) elif sample: start_date today - timedelta(days2) else: # set default to a year in the past. if we dont set a start date, we might make too many # requests to amplitude. start_date today - timedelta(days365) if sample: end_date start_date timedelta(hours1) elif end_date: if start_date end_date: return None, None if offset is not None: end_date start_date else: end_date today timedelta(hours23) if type(start_date) is str: start_date dateutil.parser.parse(start_date) if type(end_date) is str: end_date dateutil.parser.parse(end_date) return start_date, end_date各参数的实际语义可以归纳为start_date必填且优先任何情况下只要传入start_date就以它为准未传时按sample与否分别回退到「两天前」或「一年前」。源码注释特别说明默认回退到一年前是为了避免在未指定起始时间时向 Amplitude 发出过多请求。end_date可选不传时默认取「今天 23:00」传入且早于start_date时直接返回(None, None)上层load()检测到后会记录No start date and no end date.错误并返回空列表。offset小时偏移传入后会在start_date上增加offset小时并将end_date收窄为与start_date相同即只拉取一个时间点附近的数据适用于按小时回溯的增量场景。sample采样模式起始时间固定为两天前时间窗口为起始时间后的 1 小时用于快速验证连通性。数据源的test_connection()即用「今天到今天」的窗口调用load()完成连通性检查见下文。start_date/end_date也接受 ISO 字符串会自动经dateutil.parser.parse转成datetime对象这与 Mage 数据源运行时变量传入字符串格式的行为保持一致。底层实现Export API 请求、分片拉取与 zip/gzip 解压load()的主流程connections/amplitude/init.py分为三步归一化日期、按 14 天分片逐段请求、汇总解压后的 JSON 行。1. 14 天分片规避 429# Use a large timedelta or else Amplitude will return a 429 error intervals date_intervals(start_date, end_date, timedelta(days14)) for sd, ed in intervals: zip_file self.__fetch_files(sd, ed) if zip_file: data self.__build_data_from_zip_file(zip_file)代码注释直接点明了原因请求窗口过大会触发 Amplitude 的 429请求过多错误因此用 utils/dates.py 中的date_intervals()把总时间范围切成最多 14 天一段的连续区间逐段请求并拼接结果。date_intervals()会按总秒数整除分片数并为余数生成最后一个补齐区间保证覆盖完整且不重叠。2. 请求 Export API 并处理失败分支def __fetch_files(self, start_date, end_date): start_date_string start_date.strftime(DATE_FORMAT) end_date_string end_date.strftime(DATE_FORMAT) ... url f{self.host}/api/2/export?start{start_date_string}end{end_date_string} response requests.get(url, auth(self.api_key, self.secret_key)) if response.status_code 200: return io.BytesIO(response.content) elif response.status_code 403 or response.status_code 500: self.error(Failed to fetch files., ...) else: self.error(Failed to fetch files., ...) return None时间戳统一格式化为DATE_FORMAT %Y%m%dT%H例如20150201T0请求 URL 为{host}/api/2/export?start...end...。只有 HTTP 200 才会进入解压流程403鉴权失败与 5xx服务端错误会记录带reason与status_code的错误日志其余状态码也会记录错误并返回None上游对None直接跳过保证单段失败不会中断整个加载。3. 双层压缩包解压与 JSON 展平Amplitude Export API 返回的内容是 zip 压缩包内部可能再嵌套.gz文件。__build_data_from_zip_file()的处理逻辑是用zipfile.ZipFile打开外层 zip遍历内部文件名若文件名含.gz用gzip.GzipFile直接解压并读取 UTF-8 文本否则尝试把内层文件再当作 zip 打开zipfile.ZipFile(file1)继续解压其中嵌套的.gz文件捕获zipfile.BadZipFile并记录异常日志将拼接后的多行文本按换行符切分逐行json.loads解析为字典再经 utils/dictionary.py 的flatten()把事件内嵌的event_properties、user_properties等嵌套对象展平为父键_子键形式的一维键便于后续写入目标数据库。flatten()至多展开三层嵌套k1_k2_k3、k1_k2或原键。因此load()返回的每条记录都是一个扁平的字典这正是它可以直接被json.dumps打印、也可以被下游同步逻辑直接消费的原因。作为 Mage 数据源的使用增量复制与事件 Schema连接器本身是底层数据访问层真正面向数据集成管道的是 sources/amplitude/init.py 中的Amplitude(Source)数据源。它在初始化时直接复用连接器self.connection AmplitudeConnection( self.config[api_key], self.config[secret_key], hostself.config.get(host), )其增量与同步行为要点如下增量复制模式get_forced_replication_method()固定返回REPLICATION_METHOD_INCREMENTAL即 Amplitude 流只支持增量同步不做全量覆盖。复制键与主键由 constants.py 定义事件流的合法复制键为event_time与uuid主键属性为uuid。增量同步时调度框架会根据书签bookmark跳过已同步的事件。运行时日期回退未显式传入start_date/end_date时默认取「昨天到今天」today - timedelta(days1)到today传入查询参数则优先使用查询中的日期。连通性测试test_connection()调用self.connection.load(start_datetoday, end_datetoday)用一个零跨度窗口验证凭据与网络可用性这一方法会被 Mage 的「测试连接」功能触发。事件流的字段由 schemas/events.json 定义包含 40 个字段主要分几类事件标识event_id、event_type、amplitude_event_type、uuid、$insert_id、$schema时间戳event_time、client_event_time、client_upload_time、processed_time、server_received_time、server_upload_time、user_creation_time均声明为date-time格式用户与设备画像user_id、device_id、amplitude_id、session_id、os_name、os_version、platform、device_family、device_model、device_type、device_brand、device_carrier、device_manufacturer、library、start_version、version_name地理信息city、country、region、dma、language、location_lat、location_lng、ip_address业务属性event_properties、user_properties、group_properties、groups、data、paying、sample_rate、is_attribution_event、adid、idfa、amplitude_attribution_ids、app。Schema 中嵌套对象如event_properties的属性体为空实际内容在加载时由flatten()展开为动态列这种「宽表」结构便于直接落库后按事件属性做分析。通过运行时变量自定义拉取窗口在 Mage 的数据集成管道中调度触发器会自动把起始日期与结束日期传给 Amplitude 数据源例如每日运行的管道每次运行时起始日期为 1 天前、结束日期为当天对应源码中today - timedelta(days1)与today的默认值。如果需要硬编码自定义窗口可在管道变量中添加两个运行时变量参见 sources/amplitude 文档键说明示例值_start_date从 Amplitude 开始拉取事件的日期2022-10-01_end_date停止拉取事件的日期2022-10-03对应的查询模板见 templates/query.json{ start_date: , end_date: }Mage 数据源在解析查询参数时会通过query.get(start_date)、query.get(end_date)读取这两个值并经dateutil.parser.parse转为datetime后交给连接器对应load_data()中的日期覆盖逻辑。因此只要在管道运行时变量中提供_start_date/_end_date即可覆盖调度器自动推算的默认窗口。实践要点与注意事项综合源码实现使用 Amplitude 连接器时有几个值得注意的工程细节窗口越大越容易触发限流Export API 对单次请求的数据量有限制连接器内部已经按 14 天切片若自行调用load()且跨越极大时间范围仍会生成多次请求需留意 Amplitude 侧的配额与 429 返回。连接器的处理方式是按段静默跳过失败段并累计已成功数据不会因为单段失败而整体中断。鉴权失败与 5xx 会被明确记录403 与 500 以上的状态码会连同reason、status_code写入错误日志便于排查凭据过期、IP 白名单或服务端故障其余非 200 状态码同样记录日志。嵌套对象会被展平事件里的event_properties、user_properties等对象字段经flatten()变为event_properties_xxx形式的平铺列目标表的列名以父键_子键命名入库前请先确认目标数据库对动态列的支持。EU 数据驻留若 Amplitude 项目位于欧盟区域必须通过host配置https://analytics.eu.amplitude.com否则请求会打到标准服务器导致数据缺失或鉴权失败。增量书签依赖event_time/uuid增量同步的书签基于这两个复制键事件的event_time需要与 Amplitude 服务端时间一致否则可能出现重复或漏拉。本文涉及的全部实现细节均可在仓库对应文件中直接查看连接器主体 connections/amplitude/init.py、日期工具 connections/amplitude/utils.py、连接器基类 connections/base.py、数据源集成 sources/amplitude/init.py、事件字段定义 schemas/events.json以及官方数据源配置说明 docs/data-integrations/sources/amplitude.mdx。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐HYG-Database许可证变更从CC BY-SA 2.5到4.0的重要变化HYG Database许可证变更从CC BY SA 2.5到4.0的重要变化 HYG Database是一个重要的恒星数据库项目近期其许可证从CC BY数据工程数据编排ETL任务调度批处理流处理数据集成后端前端SeaTunnel Qdrant Source 连接器完全指南从向量数据库读取数据的配置、原理与实战SeaTunnel Qdrant Source 连接器完全指南从向量数据库读取数据的配置、原理与实战 Qdrant Source 是 SeaTunnel 连接数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel Zendesk Source 连接器实战从 Zendesk REST API 拉取工单数据到数据管道SeaTunnel Zendesk Source 连接器实战从 Zendesk REST API 拉取工单数据到数据管道 本文以 SeaTunnel 的 Ze数据集成ETL大数据批处理流处理变更数据捕获上一篇f1viewer安全机制跨平台凭证存储的实现与最佳实践下一篇解决LANShare传输问题常见错误排查与网络配置优化指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
