数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载本指南以开源项目 mage-ai 的mage_integrations模块为对象围绕其根目录 README.md 中记录的本地调试与端到端测试流程展开。你将掌握如何在开发容器中搭建数据集成调试环境、用 CLI 手工执行 TapSource 端与 TargetDestination 端的发现discover、取 Schema、同步sync、计数与查询最终在 Mage UI 中完成真实的数据集成 Pipeline 验证。全文配套仓库源码路径便于按图索骥深入实现。mage_integrations 模块全景mage_integrations是 mage-ai 中负责数据接入与输出的独立子项目目录布局围绕 Singer 协议SCHEMA / RECORD / STATE 三种消息组织mage_integrations/mage_integrations/sources/几十种数据源tap的实现如 GitHub、Stripe、MongoDB、PostgreSQL、S3、Salesforce 等。每个集成包含__init__.pySource 子类、templates/config.json配置模板、schemas/可选JSON Schema 定义以及各自的README.mdmage_integrations/mage_integrations/destinations/目标端target实现如 PostgreSQL、Snowflake、BigQuery、S3、Kafka 等同样包含templates/config.jsonmage_integrations/mage_integrations/connections/与各类数据库/云服务建立底层连接的工具供 sources 与 destinations 复用mage_integrations/mage_integrations/tests/针对各集成的单元测试。Source 端所有集成都继承自 sources/base.py 的Source类其process()方法base.py#L190-L284定义了整体调度逻辑先判断test_connection、load_sample_data、discover_mode、count_records_mode、show_templates等模式最后进入sync(catalog)。Destination 端则继承自 destinations/base.py 的Destination抽象类。README 中的每一段命令行操作本质上都是在直接驱动这套基类流程。环境准备进入开发容器并安装本地包1. 拉取仓库并启动开发容器git clone mage-ai 仓库地址 cd mage-ai ./scripts/dev.sh [PROJECT_NAME]scripts/dev.sh 会启动 Mage 开发用的 Docker 容器使得对源码的改动可以实时生效。启动成功后在另一个终端进入容器docker exec -it mage-ai-server-1 bash2. 卸载旧包并本地安装新包容器内默认安装了发布版的mage-integrations包为了调试仓库内的源码需要先卸载再以本地源码方式安装pip3 uninstall -y mage-integrations cd mage_integrations/注意本地源码调试与重新安装给 UI 使用是两个阶段。开发阶段直接以源码运行即可当需要让 Mage UI 感知到改动时再在mage_integrations/目录下执行pip install -U mage_integrations/README 的 Test in the Mage UI 一节对此有明确要求。该包的工程化配置见 mage_integrations/setup.py 与 mage_integrations/pyproject.toml。创建本地测试文件同步测试需要 5 个测试文件README 给出的初始化命令如下touch ./mage_integrations/TEST_CATALOG.json touch ./mage_integrations/TEST_CONFIG_S.json touch ./mage_integrations/TEST_CONFIG_D.json touch ./mage_integrations/TEST_STATE.json touch ./mage_integrations/TEST_OUTPUT echo {} ./mage_integrations/TEST_STATE.json这会在mage_integrations/目录下生成文件用途TEST_CATALOG.json记录选中的 stream 及其 JSON Schema、metadata含 selected 标记TEST_CONFIG_S.jsonSource源端配置如 GitHub 的 access_token、repositoryTEST_CONFIG_D.jsonDestination目标端配置如 PostgreSQL 的连接信息TEST_STATE.json同步断点状态bookmarks增量同步时由命令逐行写入TEST_OUTPUT无扩展名的文本输出文件用来承接同步产生的记录TEST_STATE.json必须先初始化为空 JSON 对象{}因为 Source 端在读取 state 时会执行load_json并调用check_config校验见 sources/utils.py#L264-L316。配置 SourceTEST_CONFIG_S.jsonSource 配置模板统一存放于mage_integrations/mage_integrations/sources/[INTEGRATION]/templates/config.json。以 GitHub 为例模板内容为{ access_token: abcdefghijklmnopqrstuvwxyz1234567890ABCD, repository: mage-ai/mage-ai, start_date: 2021-01-01T00:00:00Z, request_timeout: 300, base_url: https://api.github.com }将模板内容填入TEST_CONFIG_S.json并替换为真实值。GitHub 集成各参数的语义可对照 sources/github/README.md 的配置表Key说明示例值是否必填access_token有仓库访问权限的 GitHub Personal Access Tokenabcdefghijklmnopqrstuvwxyz1234567890ABCD是repository仓库路径github.com之后的部分mage-ai/mage-ai是start_date同步起始时间ISO 8601 格式2021-01-01T00:00:00Z是request_timeout请求超时秒数300是base_urlAPI 基础地址位于 repository 之前https://api.github.com是这些配置会被 sources/github/init.py 的Github类读取discover()用它构造GithubClient拉取目录sync()用它驱动同步test_connection()则调用client.verify_access_for_repo()校验令牌对仓库的访问权限。需要说明的是每个集成都有自己的配置模板与必填项新建其他集成时应以其templates/config.json与 README 为准。发现 Streams--discover --discover_streams填入配置后先用发现模式列出源端可供同步的 streampython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --discover_streams以 GitHub 为例输出是一份 stream 列表[ { stream: commits, tap_stream_id: commits }, { stream: comments, tap_stream_id: comments }, ... ]底层实现中--discover_streams与--discover组合会走 base.py#L222-L230 的分支discover_streams_mode为真时直接调用self.discover_streams()并将结果 dump 到 stdout该方法在 base.py#L150-L160 中定义返回stream与tap_stream_id构成的字典列表。GitHub 集成支持的 stream 可在其 README 中看到全量清单包括 Assignees、Collaborators、Commits、Commit Comments、Events、Issues、Issue Events、Issue Milestones、Projects、Project Cards、Project Columns、Pull Requests、PR Commits、Releases、Comments、Reviews、Review Comments、Stargazers、Teams、Team Members、Team Memberships 等。获取 Stream 的 Schema 并生成 Catalog挑选若干感兴趣的 stream用--selected_streams指定以 JSON 字符串列表形式传入把输出重定向到TEST_CATALOG.jsonpython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --selected_streams SCHEMAS mage_integrations/TEST_CATALOG.json其中SCHEMAS是占位符应替换为字符串形式的 stream 列表。例如 GitHub 源只取commitspython3 mage_integrations/sources/github/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --selected_streams [commits] mage_integrations/TEST_CATALOG.json这里--discover触发 discover 模式、--selected_streams则传入过滤列表。源码中discover(streams)base.py#L133-L148会对每个 stream 调用build_catalog_entry()后者通过get_standard_metadatasources/utils.py#L20-L47生成 metadata主键写入table-key-properties复制方法写入forced-replication-method合法复制键写入valid-replication-keys字段级inclusion根据是否为主键/复制键自动标记为automatic或available。手动启用 Schemaselected 标记build_catalog_entry()的一个重要细节是新生成的 catalog 中 stream 级selected默认是falsebase.py#L576-L578。因此 README 要求手动在TEST_CATALOG.json中为每个 stream 的顶层 metadata 加入selected: true... stream: commits, metadata: [ { breadcrumb: [], metadata: { table-key-properties: [ sha ], forced-replication-method: INCREMENTAL, valid-replication-keys: updated_at, inclusion: available, selected: true } }, ...对于 SQL 类源还需要在 stream 对象中以及至少一个列上补充{ stream: commits, tap_stream_id: commits, selected: true }原因在于同步阶段sync()只会遍历catalog.get_selected_streams(self.state or {})base.py#L531未标记 selected 的 stream 会被跳过。此外列级 metadata 的inclusion: automatic字段默认即为选中状态普通列则需要显式selected: true才会出现在 RECORD 输出中process_stream中通过extract_selected_columns(stream.metadata)决定写入哪些列见 base.py#L342-L345。该流程在 Mage UI 中由界面自动完成本步只是手工复现同样的效果。如果希望以编程方式批量设置 selected 与 key_properties可以复用 sources/utils.py#L50-L106 的update_catalog_dict。执行 Stream 同步并落盘Catalog 就绪后即可执行真正的同步python3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json mage_integrations/TEST_OUTPUT命令完成后打开TEST_OUTPUT即可看到实时同步结果。这里的输出内容是按 Singer 协议组织的消息流每个 stream 先输出SCHEMA消息由process_stream()写出包含 schema、key_properties、bookmark_properties、replication_method 等随后输出若干RECORD消息由sync_stream()经write_records()逐批写出最后在复制键存在时写出STATE消息记录 bookmark相关实现见 base.py#L286-L376 与 base.py#L378-L515。测试时指定的复制键updated_at与forced-replication-method: INCREMENTAL正是驱动 bookmark 逻辑的依据。TEST_OUTPUT的这一行行 JSON 消息同时也是下一步写目标端的输入原料。写入目标数据库Destination 端调试配置 TEST_CONFIG_D.json先按目标端模板填充TEST_CONFIG_D.json。以 PostgreSQL 为例模板位于 destinations/postgresql/templates/config.json{ database: , host: , password: , port: 5432, schema: , table: , username: }填入实际的数据库名、主机、密码、端口默认 5432、schema、目标表名与用户名。目标端会依据 destinations/postgresql/init.py 的PostgreSQL类读取这些字段build_connection()用database、host、password、port、username构造连接full_table_name()用schema.table拼接目标表全名。执行写入python3 mage_integrations/destinations/postgresql/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --input_file_path mage_integrations/TEST_OUTPUT \ --debug--input_file_path指向 Source 端产生的TEST_OUTPUT--debug开启详细日志--state读取/回写同步状态。目标端命令行参数在 destinations/base.py#L71-L103 中统一注册包括--config、--config_json、--catalog_json、--debug、--input_file_path、--state、--test_connection、--show_templates等。写入过程中SQL 类目标端会依据 SCHEMA 消息自动CREATE TABLE IF NOT EXISTS并做必要的ALTER TABLE对应 destinations/postgresql/init.py 中的build_create_table_commands与build_alter_table_commands。注意此步骤需要一个可写且已有数据表的数据源作为落点。Source 到 Destination 端到端同步把 Source 命令的输出通过管道直接接到 Destination 命令的输入即可完成拉取即写入的完整链路python3 mage_integrations/sources/[SOURCE_INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json | python3 mage_integrations/destinations/[TARGET_INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --debug例如 GitHub 到 PostgreSQL 的端到端集成python3 mage_integrations/sources/github/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json | python3 mage_integrations/destinations/postgres/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --debug这是验证某集成是否可用的最快方式Source 端逐行向 stdout 输出 SCHEMA/RECORD/STATE 消息Destination 端从 stdin 逐行消费。若目标端目录名为postgresql而非postgres请以实际目录为准见 destinations/postgresql。在 Mage UI 中做端到端验证终端测试通过后需要回到 UI 验证整条数据集成链路是否在 Mage 中可用在mage_integrations/目录执行pip install -U mage_integrations/将改动构建为新的mage-integrations包使 UI 加载到最新代码打开 Mage开发模式下访问localhost:3000新建一个 Data Integration Pipeline从集成列表中选择你的 Source依次完成三项验证测试连接Test the connection查看并选择 StreamsView and select streams将某个 Stream 同步到目标端Sync one stream to a destination。若以 Pull Request 形式提交新的 tap应在 PR 描述中附上 Source 的日志与目标表内的数据截图作为证据如果该集成支持增量同步务必同时验证检查 state 是否被正确更新并正确拉取。记录计数与样例查询统计 Stream 记录数需要知道某个 stream 有多少条可同步记录时使用--count_recordspython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json \ --count_records \ --selected_streams [your_stream]--count_records对应 base.py#L231-L248 的模式分支它会先通过catalog.get_selected_streams(self.state or {})选出已标记 selected 的 stream再过滤出--selected_streams指定的流逐个调用count_records()最终以[{count: N, id: ..., stream: ...}]的 JSON 数组形式输出到 stdout。注意各集成的count_records()默认返回 0需要具体集成覆写该方法的默认实现base.py#L616-L623。带查询参数执行样例查询若要验证带过滤条件的同步使用--query_json传入查询参数以 freshdesk 为例python3 mage_integrations/sources/freshdesk/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --query_json {_end_date: null, _execution_date: 2022-11-17T21:05:53.341319, _execution_partition: 444/20221117T210443, _start_date: null, _limit: 1000, _offset: 0} \ --state mage_integrations/TEST_STATE.json--query_json会被 sources/utils.py#L280-L285 解析为字典后传入load_data()其中的_start_date、_end_date、_execution_date、_execution_partition、_limit、_offset是各集成通用的查询参数前几个用于按执行窗口与分区过滤后两个用于分页控制。具体支持哪些参数取决于对应 Source 的load_data()实现。调试要点与常见问题每次修改源码后若通过 UI 测试务必重新执行pip install -U mage_integrations/纯终端调试则直接运行对应__init__.py即可TEST_CATALOG.json中未标记selected: true的 stream 不会进入同步这是手工测试时最容易被忽略的环节增量同步是否可用取决于具体集成例如 GitHub 集成在其 README 中明确说明当前所有端点均不支持增量同步bookmark 值存在格式问题需要修改tap_github/streams.py的IncrementalStream类方可支持。调试其他集成时请先确认其 README 中关于复制方法与 bookmarks 的说明对 SQL 源而言除了 stream 级selected: true还需为至少一个列启用 selected否则extract_selected_columns提取不到任何输出列目标端需要真实的可写数据库与数据表--debug可以输出完整的建表/写入日志便于定位类型映射或连接问题所有 Source 命令的参数-c/--config、-s/--state、--catalog、--discover、--discover_streams、--count_records、--query/--query_json、--selected_streams_json、--settings、--test_connection、--load_sample_data、--show_templates等均在 sources/utils.py#L145-L318 的parse_args中定义遇到参数行为疑问可优先查阅该文件。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐Mage 数据集成 BigQuery 目标端Destination完整配置指南与源码解析Mage 数据集成 BigQuery 目标端Destination完整配置指南与源码解析 BigQuery 是 Mage 开源数据集成框架内置的 SQL 类数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage 数据集成之 Snowflake 目标Destination完整配置指南Mage 数据集成之 Snowflake 目标Destination完整配置指南 本文聚焦于 Mage 数据集成框架中 mage_integrations数据工程数据编排ETL任务调度批处理流处理数据集成后端前端MediaPipe 上 GPU 加速只需 3 步从跑通到提速的完整路径MediaPipe 上 GPU 加速只需 3 步从跑通到提速的完整路径 你配好了 CUDA、加了 configcuda 程序跑起来了可 nvidia s数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇3条命令把老视频修成高清Video2X 免费实操教程下一篇steghide开发指南贡献代码与添加新文件格式支持的方法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Mage AI 数据集成实战指南：从源码调试 Source 与 Destination 的完整开发流程
数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载本指南以开源项目 mage-ai 的mage_integrations模块为对象围绕其根目录 README.md 中记录的本地调试与端到端测试流程展开。你将掌握如何在开发容器中搭建数据集成调试环境、用 CLI 手工执行 TapSource 端与 TargetDestination 端的发现discover、取 Schema、同步sync、计数与查询最终在 Mage UI 中完成真实的数据集成 Pipeline 验证。全文配套仓库源码路径便于按图索骥深入实现。mage_integrations 模块全景mage_integrations是 mage-ai 中负责数据接入与输出的独立子项目目录布局围绕 Singer 协议SCHEMA / RECORD / STATE 三种消息组织mage_integrations/mage_integrations/sources/几十种数据源tap的实现如 GitHub、Stripe、MongoDB、PostgreSQL、S3、Salesforce 等。每个集成包含__init__.pySource 子类、templates/config.json配置模板、schemas/可选JSON Schema 定义以及各自的README.mdmage_integrations/mage_integrations/destinations/目标端target实现如 PostgreSQL、Snowflake、BigQuery、S3、Kafka 等同样包含templates/config.jsonmage_integrations/mage_integrations/connections/与各类数据库/云服务建立底层连接的工具供 sources 与 destinations 复用mage_integrations/mage_integrations/tests/针对各集成的单元测试。Source 端所有集成都继承自 sources/base.py 的Source类其process()方法base.py#L190-L284定义了整体调度逻辑先判断test_connection、load_sample_data、discover_mode、count_records_mode、show_templates等模式最后进入sync(catalog)。Destination 端则继承自 destinations/base.py 的Destination抽象类。README 中的每一段命令行操作本质上都是在直接驱动这套基类流程。环境准备进入开发容器并安装本地包1. 拉取仓库并启动开发容器git clone mage-ai 仓库地址 cd mage-ai ./scripts/dev.sh [PROJECT_NAME]scripts/dev.sh 会启动 Mage 开发用的 Docker 容器使得对源码的改动可以实时生效。启动成功后在另一个终端进入容器docker exec -it mage-ai-server-1 bash2. 卸载旧包并本地安装新包容器内默认安装了发布版的mage-integrations包为了调试仓库内的源码需要先卸载再以本地源码方式安装pip3 uninstall -y mage-integrations cd mage_integrations/注意本地源码调试与重新安装给 UI 使用是两个阶段。开发阶段直接以源码运行即可当需要让 Mage UI 感知到改动时再在mage_integrations/目录下执行pip install -U mage_integrations/README 的 Test in the Mage UI 一节对此有明确要求。该包的工程化配置见 mage_integrations/setup.py 与 mage_integrations/pyproject.toml。创建本地测试文件同步测试需要 5 个测试文件README 给出的初始化命令如下touch ./mage_integrations/TEST_CATALOG.json touch ./mage_integrations/TEST_CONFIG_S.json touch ./mage_integrations/TEST_CONFIG_D.json touch ./mage_integrations/TEST_STATE.json touch ./mage_integrations/TEST_OUTPUT echo {} ./mage_integrations/TEST_STATE.json这会在mage_integrations/目录下生成文件用途TEST_CATALOG.json记录选中的 stream 及其 JSON Schema、metadata含 selected 标记TEST_CONFIG_S.jsonSource源端配置如 GitHub 的 access_token、repositoryTEST_CONFIG_D.jsonDestination目标端配置如 PostgreSQL 的连接信息TEST_STATE.json同步断点状态bookmarks增量同步时由命令逐行写入TEST_OUTPUT无扩展名的文本输出文件用来承接同步产生的记录TEST_STATE.json必须先初始化为空 JSON 对象{}因为 Source 端在读取 state 时会执行load_json并调用check_config校验见 sources/utils.py#L264-L316。配置 SourceTEST_CONFIG_S.jsonSource 配置模板统一存放于mage_integrations/mage_integrations/sources/[INTEGRATION]/templates/config.json。以 GitHub 为例模板内容为{ access_token: abcdefghijklmnopqrstuvwxyz1234567890ABCD, repository: mage-ai/mage-ai, start_date: 2021-01-01T00:00:00Z, request_timeout: 300, base_url: https://api.github.com }将模板内容填入TEST_CONFIG_S.json并替换为真实值。GitHub 集成各参数的语义可对照 sources/github/README.md 的配置表Key说明示例值是否必填access_token有仓库访问权限的 GitHub Personal Access Tokenabcdefghijklmnopqrstuvwxyz1234567890ABCD是repository仓库路径github.com之后的部分mage-ai/mage-ai是start_date同步起始时间ISO 8601 格式2021-01-01T00:00:00Z是request_timeout请求超时秒数300是base_urlAPI 基础地址位于 repository 之前https://api.github.com是这些配置会被 sources/github/init.py 的Github类读取discover()用它构造GithubClient拉取目录sync()用它驱动同步test_connection()则调用client.verify_access_for_repo()校验令牌对仓库的访问权限。需要说明的是每个集成都有自己的配置模板与必填项新建其他集成时应以其templates/config.json与 README 为准。发现 Streams--discover --discover_streams填入配置后先用发现模式列出源端可供同步的 streampython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --discover_streams以 GitHub 为例输出是一份 stream 列表[ { stream: commits, tap_stream_id: commits }, { stream: comments, tap_stream_id: comments }, ... ]底层实现中--discover_streams与--discover组合会走 base.py#L222-L230 的分支discover_streams_mode为真时直接调用self.discover_streams()并将结果 dump 到 stdout该方法在 base.py#L150-L160 中定义返回stream与tap_stream_id构成的字典列表。GitHub 集成支持的 stream 可在其 README 中看到全量清单包括 Assignees、Collaborators、Commits、Commit Comments、Events、Issues、Issue Events、Issue Milestones、Projects、Project Cards、Project Columns、Pull Requests、PR Commits、Releases、Comments、Reviews、Review Comments、Stargazers、Teams、Team Members、Team Memberships 等。获取 Stream 的 Schema 并生成 Catalog挑选若干感兴趣的 stream用--selected_streams指定以 JSON 字符串列表形式传入把输出重定向到TEST_CATALOG.jsonpython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --selected_streams SCHEMAS mage_integrations/TEST_CATALOG.json其中SCHEMAS是占位符应替换为字符串形式的 stream 列表。例如 GitHub 源只取commitspython3 mage_integrations/sources/github/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --discover \ --selected_streams [commits] mage_integrations/TEST_CATALOG.json这里--discover触发 discover 模式、--selected_streams则传入过滤列表。源码中discover(streams)base.py#L133-L148会对每个 stream 调用build_catalog_entry()后者通过get_standard_metadatasources/utils.py#L20-L47生成 metadata主键写入table-key-properties复制方法写入forced-replication-method合法复制键写入valid-replication-keys字段级inclusion根据是否为主键/复制键自动标记为automatic或available。手动启用 Schemaselected 标记build_catalog_entry()的一个重要细节是新生成的 catalog 中 stream 级selected默认是falsebase.py#L576-L578。因此 README 要求手动在TEST_CATALOG.json中为每个 stream 的顶层 metadata 加入selected: true... stream: commits, metadata: [ { breadcrumb: [], metadata: { table-key-properties: [ sha ], forced-replication-method: INCREMENTAL, valid-replication-keys: updated_at, inclusion: available, selected: true } }, ...对于 SQL 类源还需要在 stream 对象中以及至少一个列上补充{ stream: commits, tap_stream_id: commits, selected: true }原因在于同步阶段sync()只会遍历catalog.get_selected_streams(self.state or {})base.py#L531未标记 selected 的 stream 会被跳过。此外列级 metadata 的inclusion: automatic字段默认即为选中状态普通列则需要显式selected: true才会出现在 RECORD 输出中process_stream中通过extract_selected_columns(stream.metadata)决定写入哪些列见 base.py#L342-L345。该流程在 Mage UI 中由界面自动完成本步只是手工复现同样的效果。如果希望以编程方式批量设置 selected 与 key_properties可以复用 sources/utils.py#L50-L106 的update_catalog_dict。执行 Stream 同步并落盘Catalog 就绪后即可执行真正的同步python3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json mage_integrations/TEST_OUTPUT命令完成后打开TEST_OUTPUT即可看到实时同步结果。这里的输出内容是按 Singer 协议组织的消息流每个 stream 先输出SCHEMA消息由process_stream()写出包含 schema、key_properties、bookmark_properties、replication_method 等随后输出若干RECORD消息由sync_stream()经write_records()逐批写出最后在复制键存在时写出STATE消息记录 bookmark相关实现见 base.py#L286-L376 与 base.py#L378-L515。测试时指定的复制键updated_at与forced-replication-method: INCREMENTAL正是驱动 bookmark 逻辑的依据。TEST_OUTPUT的这一行行 JSON 消息同时也是下一步写目标端的输入原料。写入目标数据库Destination 端调试配置 TEST_CONFIG_D.json先按目标端模板填充TEST_CONFIG_D.json。以 PostgreSQL 为例模板位于 destinations/postgresql/templates/config.json{ database: , host: , password: , port: 5432, schema: , table: , username: }填入实际的数据库名、主机、密码、端口默认 5432、schema、目标表名与用户名。目标端会依据 destinations/postgresql/init.py 的PostgreSQL类读取这些字段build_connection()用database、host、password、port、username构造连接full_table_name()用schema.table拼接目标表全名。执行写入python3 mage_integrations/destinations/postgresql/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --input_file_path mage_integrations/TEST_OUTPUT \ --debug--input_file_path指向 Source 端产生的TEST_OUTPUT--debug开启详细日志--state读取/回写同步状态。目标端命令行参数在 destinations/base.py#L71-L103 中统一注册包括--config、--config_json、--catalog_json、--debug、--input_file_path、--state、--test_connection、--show_templates等。写入过程中SQL 类目标端会依据 SCHEMA 消息自动CREATE TABLE IF NOT EXISTS并做必要的ALTER TABLE对应 destinations/postgresql/init.py 中的build_create_table_commands与build_alter_table_commands。注意此步骤需要一个可写且已有数据表的数据源作为落点。Source 到 Destination 端到端同步把 Source 命令的输出通过管道直接接到 Destination 命令的输入即可完成拉取即写入的完整链路python3 mage_integrations/sources/[SOURCE_INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json | python3 mage_integrations/destinations/[TARGET_INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --debug例如 GitHub 到 PostgreSQL 的端到端集成python3 mage_integrations/sources/github/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json | python3 mage_integrations/destinations/postgres/__init__.py \ --config mage_integrations/TEST_CONFIG_D.json \ --state mage_integrations/TEST_STATE.json \ --debug这是验证某集成是否可用的最快方式Source 端逐行向 stdout 输出 SCHEMA/RECORD/STATE 消息Destination 端从 stdin 逐行消费。若目标端目录名为postgresql而非postgres请以实际目录为准见 destinations/postgresql。在 Mage UI 中做端到端验证终端测试通过后需要回到 UI 验证整条数据集成链路是否在 Mage 中可用在mage_integrations/目录执行pip install -U mage_integrations/将改动构建为新的mage-integrations包使 UI 加载到最新代码打开 Mage开发模式下访问localhost:3000新建一个 Data Integration Pipeline从集成列表中选择你的 Source依次完成三项验证测试连接Test the connection查看并选择 StreamsView and select streams将某个 Stream 同步到目标端Sync one stream to a destination。若以 Pull Request 形式提交新的 tap应在 PR 描述中附上 Source 的日志与目标表内的数据截图作为证据如果该集成支持增量同步务必同时验证检查 state 是否被正确更新并正确拉取。记录计数与样例查询统计 Stream 记录数需要知道某个 stream 有多少条可同步记录时使用--count_recordspython3 mage_integrations/sources/[INTEGRATION]/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --state mage_integrations/TEST_STATE.json \ --count_records \ --selected_streams [your_stream]--count_records对应 base.py#L231-L248 的模式分支它会先通过catalog.get_selected_streams(self.state or {})选出已标记 selected 的 stream再过滤出--selected_streams指定的流逐个调用count_records()最终以[{count: N, id: ..., stream: ...}]的 JSON 数组形式输出到 stdout。注意各集成的count_records()默认返回 0需要具体集成覆写该方法的默认实现base.py#L616-L623。带查询参数执行样例查询若要验证带过滤条件的同步使用--query_json传入查询参数以 freshdesk 为例python3 mage_integrations/sources/freshdesk/__init__.py \ --config mage_integrations/TEST_CONFIG_S.json \ --catalog mage_integrations/TEST_CATALOG.json \ --query_json {_end_date: null, _execution_date: 2022-11-17T21:05:53.341319, _execution_partition: 444/20221117T210443, _start_date: null, _limit: 1000, _offset: 0} \ --state mage_integrations/TEST_STATE.json--query_json会被 sources/utils.py#L280-L285 解析为字典后传入load_data()其中的_start_date、_end_date、_execution_date、_execution_partition、_limit、_offset是各集成通用的查询参数前几个用于按执行窗口与分区过滤后两个用于分页控制。具体支持哪些参数取决于对应 Source 的load_data()实现。调试要点与常见问题每次修改源码后若通过 UI 测试务必重新执行pip install -U mage_integrations/纯终端调试则直接运行对应__init__.py即可TEST_CATALOG.json中未标记selected: true的 stream 不会进入同步这是手工测试时最容易被忽略的环节增量同步是否可用取决于具体集成例如 GitHub 集成在其 README 中明确说明当前所有端点均不支持增量同步bookmark 值存在格式问题需要修改tap_github/streams.py的IncrementalStream类方可支持。调试其他集成时请先确认其 README 中关于复制方法与 bookmarks 的说明对 SQL 源而言除了 stream 级selected: true还需为至少一个列启用 selected否则extract_selected_columns提取不到任何输出列目标端需要真实的可写数据库与数据表--debug可以输出完整的建表/写入日志便于定位类型映射或连接问题所有 Source 命令的参数-c/--config、-s/--state、--catalog、--discover、--discover_streams、--count_records、--query/--query_json、--selected_streams_json、--settings、--test_connection、--load_sample_data、--show_templates等均在 sources/utils.py#L145-L318 的parse_args中定义遇到参数行为疑问可优先查阅该文件。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐Mage 数据集成 BigQuery 目标端Destination完整配置指南与源码解析Mage 数据集成 BigQuery 目标端Destination完整配置指南与源码解析 BigQuery 是 Mage 开源数据集成框架内置的 SQL 类数据工程数据编排ETL任务调度批处理流处理数据集成后端前端Mage 数据集成之 Snowflake 目标Destination完整配置指南Mage 数据集成之 Snowflake 目标Destination完整配置指南 本文聚焦于 Mage 数据集成框架中 mage_integrations数据工程数据编排ETL任务调度批处理流处理数据集成后端前端MediaPipe 上 GPU 加速只需 3 步从跑通到提速的完整路径MediaPipe 上 GPU 加速只需 3 步从跑通到提速的完整路径 你配好了 CUDA、加了 configcuda 程序跑起来了可 nvidia s数据工程数据编排ETL任务调度批处理流处理数据集成后端前端上一篇3条命令把老视频修成高清Video2X 免费实操教程下一篇steghide开发指南贡献代码与添加新文件格式支持的方法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
