数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载Knowi 是一款 BI 与数据可视化平台其 Management API 提供了对 Dashboard仪表盘和 Widget组件等元数据的访问能力。Mage 通过内置的 Knowi 数据源Source可以基于 Bearer Token 认证将 Knowi 中的仪表盘与组件元数据抽取进数据管道。本指南以开源仓库中 Knowi 数据源文档 为骨架结合 源码实现 与 Stream 定义完整讲解配置参数、access_token获取方式、可用数据流以及底层抽取与重试机制帮助你在 Mage 中快速落地 Knowi 数据集成。一、Knowi 数据源概览Knowi 数据源是 Mage 内置的众多数据集成 Source 之一在mage_ai/data_integrations/sources/constants.py中与 Airtable、Salesforce、Stripe 等一同登记为官方支持的接入源。整个 Source 的实现位于入口与抽取逻辑mage_integrations/mage_integrations/sources/knowi/init.pyAPI 客户端mage_integrations/mage_integrations/sources/knowi/client.py数据流Stream定义mage_integrations/mage_integrations/sources/knowi/streams.pySchema 加载mage_integrations/mage_integrations/sources/knowi/schema.py字段级 Schemaschemas/dashboards.json 与 schemas/widgets.json从源码结构看该 Source 遵循 Singer 协议的 tap 风格通过load_data方法按 Stream 逐个拉取记录每个 Stream 决定自身的复制方法Replication Method、主键Key Properties与合法复制键Valid Replication Keys。二、配置参数详解配置 Knowi 数据源时需要提供以下凭证与参数。下表完整继承自 README 配置表并与 templates/config.json 中声明的字段一一对应Key描述示例值必填access_token用于认证 Knowi 账号的访问令牌Bearer Tokenabcdefg123456是REQUIREDrequest_timeout请求超时时间秒默认 300 秒300否user_agent随请求头发送的用户代理User Agent字符串my-app-v1.0否start_date过滤结果使用增量同步Incremental Sync时仅拉取在start_date之后更新的记录2023-01-01否在 Mage 的数据集成配置界面中这些参数会被写入连接配置最终作为config字典传给 Source 实例。templates/config.json提供了标准的配置模板骨架{ access_token: , start_date: null, request_timeout: null, user_agent: null }1.access_token必填Knowi API 的认证凭据这是唯一必填参数。Knowi 采用 Bearer Token 认证客户端在每次请求的 HTTP 头中携带Authorization: Bearer access_token。获取方式请遵循 Knowi 官方文档中Management API 的 API Authentication一节给出的指引在 Knowi 账号的 API 管理页面中创建并获取访问令牌然后在 Mage 的 Knowi Source 配置中将该令牌填入access_token字段。从源码看KnowiClient 构造函数 会接收access_token并保存如果该值为Nonecheck_access_token 会直接抛出Error: Missing access_token.异常。同时当 Knowi 返回 HTTP 401 且错误码中包含access_token时客户端会明确提示当前令牌已过期或无效要求重新认证以生成新的令牌后继续抽取见 raise_for_error。安全建议将access_token视为敏感凭据在 Mage 中可通过密钥管理Secrets机制引用环境变量避免明文写入代码仓库。2.request_timeout可选请求超时控制指定单个 HTTP 请求的超时秒数默认值为 300 秒。在 KnowiClient 初始化逻辑 中超时值按如下规则解析若config_request_timeout存在且float(config_request_timeout)为真即非 0、非空字符串、非0则采用该值否则回退到模块级常量REQUEST_TIMEOUT 300。也就是说传入0、0或空值都会被当作未配置最终使用 300 秒默认值。该超时值随后在check_access_token与request方法中以timeoutself.__request_timeout传入底层requests.Session调用。3.user_agent可选请求标识自定义的 User Agent 字符串会写入每次请求的User-Agent请求头便于 Knowi 侧识别调用方应用例如my-app-v1.0。源码中只要user_agent非空就会在 check_access_token 与 request 的请求头中注入。4.start_date可选增量同步的起始时间点当 Stream 采用增量复制INCREMENTAL时start_date用于过滤记录仅同步该时间点之后更新的数据。在 load_data 中可以看到具体逻辑从 bookmark 中优先读取updated_at作为断点若 bookmark 不存在则回退到配置中的start_date。bookmark 为整数时按 Unix 时间戳解析否则用singer.utils.strptime_to_utc解析为 UTC 时间。格式建议为YYYY-MM-DD如2023-01-01。三、如何获取access_token根据 README 与官方文档站对应页面docs/data-integrations/sources/knowi.mdx的说明登录 Knowi 账号进入Management API相关的 API 认证API Authentication页面按 Knowi 官方指引创建 API Token将生成的令牌复制到 Mage 中 Knowi Source 的access_token配置项。需要注意的是Knowi 的访问令牌存在有效期/失效机制。源码在捕获 401 响应时会记录如下日志提醒你重新认证并刷新令牌后继续同步Your API access_token is expired/invalid as per Knowis security policy. Please re-authenticate your connection to generate a new access_token and resume extraction.四、可用的数据流Streams与复制方式Knowi 数据源共暴露 4 个 Stream全部定义在 streams.py 的STREAMS字典中Stream ID复制方式主键说明dashboard_listFULL_TABLEid仪表盘列表仅作为父 Stream 供子 Stream 消费不参与同步dashboardsFULL_TABLEid仪表盘详情逐条拉取每个仪表盘对象widget_listFULL_TABLEid组件列表仅作为父 Stream 供子 Stream 消费不参与同步widgetsFULL_TABLEid组件详情逐条拉取每个组件对象值得注意的实现细节父子 Stream 设计DashboardList与WidgetList的to_replicate False它们不会作为独立表被同步而是通过 get_parent_data 向子 Stream 提供父级 ID 列表。例如 Dashboards.get_records 先拿到所有dashboard_id再逐个请求dashboards/{id}组装详情。复制方式当前 4 个 Stream 均为FULL_TABLE全量复制。这与start_date的语义对应——start_date仅在 Stream 的复制方式为 INCREMENTAL 时生效全量同步下bookmark_datetime会被置为None见 load_data。响应数据键Knowi API 列表类响应统一放在list键下default_data_key list详情类数据则使用各自的data_key如dashboards、widgets。每个 Stream 的主键、复制键与复制方式会通过 get_forced_replication_method、get_table_key_properties、get_valid_replication_keys暴露给 Mage 编排层用于决定目标表的 Schema、主键与增量策略。字段级 Schema 则由 schema.py 在 Discovery 阶段从schemas/*.json加载并通过 Singer metadata 标注key_properties、valid_replication_keys与replication_method。以dashboards为例schemas/dashboards.json 声明的字段包括id整数主键、name、url、createdDate、lastModDt、lastAccessDt、userId、displayOrder、favorite、locked、accessLevel等足以支撑仪表盘清单的元数据同步场景。五、底层请求机制认证、限流与错误处理1. API 基地址与请求封装KnowiClient 使用https://knowi.com/api/1.0作为基础地址基于requests.Session管理连接。request方法统一完成三件事认证头注入Authorization: Bearer ...、Accept/Content-Type 头设置、以及响应状态码校验。非 200 响应会进入 raise_for_error 解析错误详情5xx 状态码则直接抛出Server5xxError以触发上层重试。2. 限流Rate Limit客户端通过utils.ratelimit(1000, 60)装饰器对check_access_token与request两个入口施加限流——即每 60 秒最多发起 1000 次请求。utils.ratelimit来自 Singer 的singer.utils若触发限流会抛出异常并由上层按照重试策略处理。3. 错误码到异常的类型化映射client.py维护了一张完整的 HTTP 错误码到异常类的映射表状态码异常类400KnowiBadRequestError401KnowiUnauthorizedError402KnowiPaymentRequiredError403KnowiForbiddenError404KnowiNotFoundError405KnowiMethodNotAllowedError406KnowiNotAcceptableError408KnowiRequestTimeoutError409KnowiUserConflictError415KnowiUnsupportedMediaTypeError422KnowiUnprocessableEntityError423KnowiScrollExistsError特殊处理scroll_exists错误码500KnowiInternalServiceError其他KnowiError兜底这些异常类统一继承自KnowiError其中Server5xxError、Server429Error通常会被上层认定为可重试错误从而实现失败请求的自动重试。4. 连接测试Mage 在保存/测试连接时会调用 test_connection其内部实例化客户端并执行check_access_token()。该方法请求https://knowi.com/api/1.0/dashboards这个轻量端点只要返回 200 且响应 JSON 中包含list键即判定令牌有效、连接成功否则记录Error status_code ...并抛出对应异常。这也是你在界面点Test Connection时背后真正执行的校验逻辑。六、在 Mage 中接入 Knowi 的实践步骤1. 创建数据集成管道在 Mage 界面新建一个Data Integration类型管道选择数据源Source为Knowi对应mage_ai/data_integrations/sources/constants.py中登记的Knowi项填入上文所述的 4 个配置参数其中access_token必填点击测试连接Mage 会调用test_connection校验令牌与网络连通性选择要同步的 Streamdashboards、widgets等配置目标Destination即可启动抽取。2. 命令行验证可选Mage 的 Source 入口支持以脚本方式运行python mage_integrations/mage_integrations/sources/knowi/__init__.py --config config.json --state state.json其中config.json的结构即 templates/config.json 中声明的字段。实际抽取由load_data驱动每个 Stream 产生的记录以单条列表形式逐条 yield见init.py 第 45-46 行供下游目标按批次写入。3. 运行与排障建议若同步任务在开始时失败优先检查access_token是否有效可重新测试连接若日志出现Server5xxError或 429 相关异常说明 Knowi 侧服务异常或触发限流可在重试窗口后恢复全量同步场景下start_date不参与过滤如需时间维度增量抽取需要对应的 Stream 支持 INCREMENTAL 复制方式。七、总结Knowi 数据源为 Mage 用户提供了一条将 Knowi 仪表盘与组件元数据接入数据管道的捷径。核心要点可归纳为认证access_token是唯一必填项通过 Bearer Token 完成认证令牌失效时需要重新生成配置request_timeout默认 300 秒、user_agent、start_date为可选参数语义与实现一一对应数据流dashboards与widgets两个可同步表采用全量复制父 Stream 仅用于提供对象 ID 列表健壮性底层具备类型化错误映射、限流与 5xx 重试机制并有独立的连接测试入口方便在配置阶段即验证凭证有效性。掌握了这些细节后你就可以在 Mage 中稳定地接入 Knowi 数据源并将其纳入统一的数据集成与编排流程。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐AMD量化模型生产部署终极指南容器化、监控与性能优化全流程AMD量化模型生产部署终极指南容器化、监控与性能优化全流程 在当今AI应用快速发展的时代 AMD量化模型 的生产部署已成为企业实现高效推理的关键技术。本文将数据工程数据编排ETL任务调度批处理流处理数据集成后端前端OpenMetadata Salesforce 数据库连接器配置指南认证方式、连接参数与元数据摄取全解析OpenMetadata Salesforce 数据库连接器配置指南认证方式、连接参数与元数据摄取全解析 Salesforce 作为 CRM 领域的核心平台数据目录数据血缘数据治理后端MCP 服务OpenMetadata Alation 连接器配置指南认证、后端数据库直连与元数据摄取详解OpenMetadata Alation 连接器配置指南认证、后端数据库直连与元数据摄取详解 导读 本文档面向使用 OpenMetadata 对接 Alati数据目录数据血缘数据治理后端MCP 服务上一篇探索高效实时定位FAST-LIVO开源项目详解下一篇【亲测免费】 探索Alpaca Eval一个高效、灵活的自然语言处理评估工具创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考