尧图网页设计 
使用 Google Cloud Functions 部署 dlt 数据加载管道Pipeline【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt导读本文基于 dltdata load tool官方仓库的部署指南完整讲解如何利用gcloud命令行Cloud Shell与dltCLI将 dlt 管道部署为 Google Cloud FunctionsGCF上的 HTTP 触发函数从dlt init初始化 Notion → BigQuery 管道开始到编写入口函数main.py、执行gcloud functions deploy部署、配置环境变量含 Secret Manager 方案再到监控与手动触发并进一步将函数改造为实时事件摄入 Webhook。读完本文你将掌握一套可直接落地的「dlt 管道上云」流程并理解其背后与仓库源码相关的初始化、配置加载与调度机制。前置条件与适用场景本文的部署方式以gcloud命令行Cloud Shell和 dlt CLI 为核心操作入口。在开始之前你需要具备GCP 基础能力对 Cloud Functions、IAM 与权限、GCP 服务账号Service Account有基本认识dlt 项目环境本地已安装 dlt且能够在终端执行dlt命令代码所在目录函数代码位于本地机器或云端代码仓库如 GitHub、Bitbucket的目录中部署时将基于该目录执行。适用的典型场景包括将定时或按需的批式数据管道例如 Notion 数据加载到 BigQuery以 HTTP 函数形式暴露将 Cloud Functions 作为事件摄入 Webhook对接 Slack、Discord、Stripe、PayPal 等支持 webhook 的平台实现事件数据的实时入库。1. 初始化 dlt 管道1.1 使用 dlt init 生成管道骨架在本指南中我们以 dlt 的 Notion verified source 为例目标端destination选用 BigQuery。你也可以替换为任何其他 verified source或自行编写自定义 source。在终端中执行dlt init notion bigquery命令执行完成后会在当前主目录中生成一批文件与文件夹包含管道所需的全部配置例如notion_pipeline.py可运行的管道示例脚本requirements.txt依赖清单含dlt[bigquery]等secrets.toml密钥占位文件Notion API key 等敏感信息存放处.dlt/或对应配置目录。从仓库源码看dlt init实际调用的是 dlt/_workspace/cli/_init_command.py 中的init_command()/init_pipeline_at_destination()其职责是“为source_name → destination_type脚手架化管道config、secrets、可选的示例脚本”支持eject_source将 source 源码拷贝进项目以便修改、dry_run只打印计划不落盘等参数。Notion 验证源的完整使用说明见 notion.md其中notion_databases资源用于从 Notion 数据库检索数据。1.2 编写 main.py 入口文件在主目录下新建main.py内容如下from notion_pipeline import load_databases def pipeline_notion(request): load_databases() return Pipeline run successfully!关键说明Google Cloud Functions 默认会在目录中查找main.py文件函数名pipeline_notion将作为部署时的“入口点”Entry pointrequest参数由 GCF 运行时注入当前示例未使用但保留签名以便扩展。1.3 补充依赖如管道需要额外的第三方包将其追加到dlt init生成的requirements.txt中。GCF 在部署时会根据该文件安装依赖。2. 部署 GCP Cloud Function进入main.py所在目录执行gcloud functions deploy pipeline_notion --runtime python310 \ --trigger-http --allow-unauthenticated --source . --timeout 300参数含义如下参数含义pipeline_notion部署的函数名与main.py中定义的入口函数一致--runtime python310使用 Python 3.10 运行时环境--trigger-http以 HTTP 方式触发函数--allow-unauthenticated允许未认证调用即任何人访问触发 URL 都可执行--source .将当前目录全部文件作为函数源码上传--timeout 300函数超时时间 300 秒5 分钟超时时间说明默认 5 分钟适合多数管道如果向目标端上传大量文件HTTP 函数最长可提升至 60 分钟事件驱动函数最长 10 分钟请根据实际数据量在 GCF 控制台的超时配置中调整。权限提醒你的 GCP 项目有一个与项目 ID 关联的默认服务账号请为该服务账号授予Cloud Functions Developer角色否则部署可能因权限不足失败。3. 在 Cloud Function 中配置环境变量dlt 的密钥与配置加载遵循secrets.toml/config.toml机制仓库中 dlt/common/configuration/providers/toml.py 定义了SECRETS_TOML secrets.toml。云函数环境中没有本地文件系统上的secrets.toml因此需要通过环境变量注入。共有两种方式3a. 直接在函数中声明环境变量进入 Google Cloud Functions选中已部署的函数点击EDIT切换到BUILD页签在BUILD ENVIRONMENTAL VARIABLE下点击ADD VARIABLE变量名需对应管道参数。若该参数在secrets.toml中以小写命名例如api_key则环境变量名必须大写即API_KEY填写 Notion API key 的值点击Next并部署函数。原理说明dlt 的配置注入器会尝试从环境变量读取配置项环境变量名按节与参数名大写形式组织如API_KEY对应api_key这与secrets.toml中的小写键形成映射。3b. 使用 GCP Secret Manager进入 Google Cloud Functions选中函数点击EDIT在Runtime, Build, Connections and Security Settings区域选择Security and Images Repo点击Add a secret reference选择已创建的 Secret例如notion_secret将Reference method设置为Mounted as environment variable在Environment Variable字段填写与管道参数对应的环境变量名同样需大写如API_KEY对应api_key点击DEPLOY完成部署。此后每次触发 URLHTTP 触发器都会成功执行管道权限为部署云函数所用的服务账号通常即项目默认服务账号授予Secret Manager Secret Accessor角色才能读取 Secret。4. 监控与手动触发函数部署完成后GCF 会生成一个 Trigger URL。在浏览器地址栏打开该 URL 即可手动触发函数页面返回Pipeline run successfully!说明管道成功执行数据已成功加载到目标端BigQuery若返回其他错误请结合 GCF 的日志Logs与 dlt 的 trace 信息排查。5. 将 Cloud Function 部署为 Webhook5.1 Webhook 简介Webhook 是一种应用间实时推送消息/数据的机制与需要持续轮询的传统 API 不同Webhook 在事件发生时立即将信息推送给接收方。这种事件驱动架构响应更快、更省资源也提升了系统整体性能。借助 dlt 的 Google Cloud 事件摄入 Webhook可以在 Webhook POST 请求触发的瞬间实时摄入数据并加载到目标端适用于 Slack、Discord、Stripe、PayPal 等任意支持 webhook 的平台。5.2 初始化部署登录 GCP 账号并启用 Cloud Functions API进入 Cloud Functions 页面点击Create Function配置运行环境并选择区域配置触发器类型本示例使用 HTTP并选择Allow unauthenticated invocations允许未认证调用点击Save后Next运行时环境选择Python 3.10使用以下代码设置事件摄入函数import dlt import time from google.cloud import bigquery from dlt.common import json def your_webhook(request): # Extract relevant data from the request payload data request.get_json() Event [data] pipeline dlt.pipeline( pipeline_nameplatform_to_bigquery, destinationbigquery, dataset_namewebhooks, ) pipeline.run(Event, table_namewebhook) #table_name can be customized return Event received and processed successfully.在Entry point字段将函数名设置为your_webhook在requirements.txt中声明依赖# Function dependencies, for example: # packageversion dlt dlt[bigquery]点击Deploy完成部署。完成之后该 Cloud Function 即可作为 Slack、Discord、Stripe、PayPal 等平台的 Webhook 接收端。在平台侧配置 Webhook 时请填入 Cloud Function 的Trigger URL可在 Trigger 页签中找到。5.3 手动测试 Webhook部署完成后可以发送一个手动 POST 请求来验证import requests webhook_url please set me up! # Your cloud function Trigger URL message { text: Hello, Slack!, user: dlthub, channel: dlthub } response requests.post(webhook_url, jsonmessage) if response.status_code 200: print(Message sent successfully.) else: print(Failed to send message. Error:, response.text)注意将webhook_url替换为创建函数时生成的 Trigger URL。请求返回 200 即表示函数接收并处理成功。从仓库源码看dlt.pipeline(...)会创建管道实例pipeline.run(Event, table_namewebhook)触发完整的 extract → normalize → load 流程将事件数据写入 BigQuery 的webhooks数据集dataset与webhook表。数据写入与加载的底层实现可进一步参考 dlt/pipeline/pipeline.py 与 dlt/load/load.py。6. 运行流程与底层机制小结整个方案的运行时链路可归纳为Webhook POST / HTTP GETTrigger URL │ ▼ Cloud Functions 入口函数main.py: pipeline_notion / your_webhook │ ▼ dlt pipelinepipeline.run(...) │ ▼ Extractsource 资源迭代→ Normalize数据标准化→ Load写入目标端 │ ▼ BigQuerydataset: webhooks / 默认数据集入口封装main.py是 GCF 默认查找的模块request参数携带 HTTP 请求上下文request.get_json()用于 Webhook 场景下解析 JSON 负载配置注入dlt 通过 dlt/common/configuration/accessors.py 中的配置访问器从secrets.toml、环境变量等多级 provider 解析密钥与参数云函数场景下以环境变量大写命名替代secrets.toml依赖管理requirements.txt中声明dlt与dlt[bigquery]前者为 dlt 核心后者额外安装 BigQuery 目标端所需的依赖如google-cloud-bigquery。7. 常见问题与排查建议部署时报权限错误确认服务账号已具备Cloud Functions Developer角色使用 Secret Manager 时额外确认Secret Manager Secret Accessor角色函数超时大数据量上传时在部署命令或控制台将--timeout从 300 秒调高HTTP 函数上限 60 分钟密钥读取失败检查环境变量名是否与secrets.toml中的小写键一一对应并转为大写如api_key→API_KEYWebhook 收不到数据确认平台侧配置的 Trigger URL 正确、函数允许未认证调用并查看 GCF 日志中是否出现Event received and processed successfully.。总结本文完整演示了 dlt 管道在 Google Cloud Functions 上的两种典型部署形态一是以dlt initmain.pygcloud functions deploy构建的可手动触发管道二是面向实时事件的 Webhook 摄入函数。两者的核心都是「dlt 管道 HTTP 触发器」区别仅在于入口函数的编写方式与触发方式GET 手动触发 vs POST 事件推送。借助 dlt 统一的pipeline.run()接口与 BigQuery 目标端数据从 Notion 数据库或各类平台事件到数仓的链路可在数分钟内完成上云。【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 Google Cloud Functions 部署 dlt 数据加载管道（Pipeline）
使用 Google Cloud Functions 部署 dlt 数据加载管道Pipeline【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt导读本文基于 dltdata load tool官方仓库的部署指南完整讲解如何利用gcloud命令行Cloud Shell与dltCLI将 dlt 管道部署为 Google Cloud FunctionsGCF上的 HTTP 触发函数从dlt init初始化 Notion → BigQuery 管道开始到编写入口函数main.py、执行gcloud functions deploy部署、配置环境变量含 Secret Manager 方案再到监控与手动触发并进一步将函数改造为实时事件摄入 Webhook。读完本文你将掌握一套可直接落地的「dlt 管道上云」流程并理解其背后与仓库源码相关的初始化、配置加载与调度机制。前置条件与适用场景本文的部署方式以gcloud命令行Cloud Shell和 dlt CLI 为核心操作入口。在开始之前你需要具备GCP 基础能力对 Cloud Functions、IAM 与权限、GCP 服务账号Service Account有基本认识dlt 项目环境本地已安装 dlt且能够在终端执行dlt命令代码所在目录函数代码位于本地机器或云端代码仓库如 GitHub、Bitbucket的目录中部署时将基于该目录执行。适用的典型场景包括将定时或按需的批式数据管道例如 Notion 数据加载到 BigQuery以 HTTP 函数形式暴露将 Cloud Functions 作为事件摄入 Webhook对接 Slack、Discord、Stripe、PayPal 等支持 webhook 的平台实现事件数据的实时入库。1. 初始化 dlt 管道1.1 使用 dlt init 生成管道骨架在本指南中我们以 dlt 的 Notion verified source 为例目标端destination选用 BigQuery。你也可以替换为任何其他 verified source或自行编写自定义 source。在终端中执行dlt init notion bigquery命令执行完成后会在当前主目录中生成一批文件与文件夹包含管道所需的全部配置例如notion_pipeline.py可运行的管道示例脚本requirements.txt依赖清单含dlt[bigquery]等secrets.toml密钥占位文件Notion API key 等敏感信息存放处.dlt/或对应配置目录。从仓库源码看dlt init实际调用的是 dlt/_workspace/cli/_init_command.py 中的init_command()/init_pipeline_at_destination()其职责是“为source_name → destination_type脚手架化管道config、secrets、可选的示例脚本”支持eject_source将 source 源码拷贝进项目以便修改、dry_run只打印计划不落盘等参数。Notion 验证源的完整使用说明见 notion.md其中notion_databases资源用于从 Notion 数据库检索数据。1.2 编写 main.py 入口文件在主目录下新建main.py内容如下from notion_pipeline import load_databases def pipeline_notion(request): load_databases() return Pipeline run successfully!关键说明Google Cloud Functions 默认会在目录中查找main.py文件函数名pipeline_notion将作为部署时的“入口点”Entry pointrequest参数由 GCF 运行时注入当前示例未使用但保留签名以便扩展。1.3 补充依赖如管道需要额外的第三方包将其追加到dlt init生成的requirements.txt中。GCF 在部署时会根据该文件安装依赖。2. 部署 GCP Cloud Function进入main.py所在目录执行gcloud functions deploy pipeline_notion --runtime python310 \ --trigger-http --allow-unauthenticated --source . --timeout 300参数含义如下参数含义pipeline_notion部署的函数名与main.py中定义的入口函数一致--runtime python310使用 Python 3.10 运行时环境--trigger-http以 HTTP 方式触发函数--allow-unauthenticated允许未认证调用即任何人访问触发 URL 都可执行--source .将当前目录全部文件作为函数源码上传--timeout 300函数超时时间 300 秒5 分钟超时时间说明默认 5 分钟适合多数管道如果向目标端上传大量文件HTTP 函数最长可提升至 60 分钟事件驱动函数最长 10 分钟请根据实际数据量在 GCF 控制台的超时配置中调整。权限提醒你的 GCP 项目有一个与项目 ID 关联的默认服务账号请为该服务账号授予Cloud Functions Developer角色否则部署可能因权限不足失败。3. 在 Cloud Function 中配置环境变量dlt 的密钥与配置加载遵循secrets.toml/config.toml机制仓库中 dlt/common/configuration/providers/toml.py 定义了SECRETS_TOML secrets.toml。云函数环境中没有本地文件系统上的secrets.toml因此需要通过环境变量注入。共有两种方式3a. 直接在函数中声明环境变量进入 Google Cloud Functions选中已部署的函数点击EDIT切换到BUILD页签在BUILD ENVIRONMENTAL VARIABLE下点击ADD VARIABLE变量名需对应管道参数。若该参数在secrets.toml中以小写命名例如api_key则环境变量名必须大写即API_KEY填写 Notion API key 的值点击Next并部署函数。原理说明dlt 的配置注入器会尝试从环境变量读取配置项环境变量名按节与参数名大写形式组织如API_KEY对应api_key这与secrets.toml中的小写键形成映射。3b. 使用 GCP Secret Manager进入 Google Cloud Functions选中函数点击EDIT在Runtime, Build, Connections and Security Settings区域选择Security and Images Repo点击Add a secret reference选择已创建的 Secret例如notion_secret将Reference method设置为Mounted as environment variable在Environment Variable字段填写与管道参数对应的环境变量名同样需大写如API_KEY对应api_key点击DEPLOY完成部署。此后每次触发 URLHTTP 触发器都会成功执行管道权限为部署云函数所用的服务账号通常即项目默认服务账号授予Secret Manager Secret Accessor角色才能读取 Secret。4. 监控与手动触发函数部署完成后GCF 会生成一个 Trigger URL。在浏览器地址栏打开该 URL 即可手动触发函数页面返回Pipeline run successfully!说明管道成功执行数据已成功加载到目标端BigQuery若返回其他错误请结合 GCF 的日志Logs与 dlt 的 trace 信息排查。5. 将 Cloud Function 部署为 Webhook5.1 Webhook 简介Webhook 是一种应用间实时推送消息/数据的机制与需要持续轮询的传统 API 不同Webhook 在事件发生时立即将信息推送给接收方。这种事件驱动架构响应更快、更省资源也提升了系统整体性能。借助 dlt 的 Google Cloud 事件摄入 Webhook可以在 Webhook POST 请求触发的瞬间实时摄入数据并加载到目标端适用于 Slack、Discord、Stripe、PayPal 等任意支持 webhook 的平台。5.2 初始化部署登录 GCP 账号并启用 Cloud Functions API进入 Cloud Functions 页面点击Create Function配置运行环境并选择区域配置触发器类型本示例使用 HTTP并选择Allow unauthenticated invocations允许未认证调用点击Save后Next运行时环境选择Python 3.10使用以下代码设置事件摄入函数import dlt import time from google.cloud import bigquery from dlt.common import json def your_webhook(request): # Extract relevant data from the request payload data request.get_json() Event [data] pipeline dlt.pipeline( pipeline_nameplatform_to_bigquery, destinationbigquery, dataset_namewebhooks, ) pipeline.run(Event, table_namewebhook) #table_name can be customized return Event received and processed successfully.在Entry point字段将函数名设置为your_webhook在requirements.txt中声明依赖# Function dependencies, for example: # packageversion dlt dlt[bigquery]点击Deploy完成部署。完成之后该 Cloud Function 即可作为 Slack、Discord、Stripe、PayPal 等平台的 Webhook 接收端。在平台侧配置 Webhook 时请填入 Cloud Function 的Trigger URL可在 Trigger 页签中找到。5.3 手动测试 Webhook部署完成后可以发送一个手动 POST 请求来验证import requests webhook_url please set me up! # Your cloud function Trigger URL message { text: Hello, Slack!, user: dlthub, channel: dlthub } response requests.post(webhook_url, jsonmessage) if response.status_code 200: print(Message sent successfully.) else: print(Failed to send message. Error:, response.text)注意将webhook_url替换为创建函数时生成的 Trigger URL。请求返回 200 即表示函数接收并处理成功。从仓库源码看dlt.pipeline(...)会创建管道实例pipeline.run(Event, table_namewebhook)触发完整的 extract → normalize → load 流程将事件数据写入 BigQuery 的webhooks数据集dataset与webhook表。数据写入与加载的底层实现可进一步参考 dlt/pipeline/pipeline.py 与 dlt/load/load.py。6. 运行流程与底层机制小结整个方案的运行时链路可归纳为Webhook POST / HTTP GETTrigger URL │ ▼ Cloud Functions 入口函数main.py: pipeline_notion / your_webhook │ ▼ dlt pipelinepipeline.run(...) │ ▼ Extractsource 资源迭代→ Normalize数据标准化→ Load写入目标端 │ ▼ BigQuerydataset: webhooks / 默认数据集入口封装main.py是 GCF 默认查找的模块request参数携带 HTTP 请求上下文request.get_json()用于 Webhook 场景下解析 JSON 负载配置注入dlt 通过 dlt/common/configuration/accessors.py 中的配置访问器从secrets.toml、环境变量等多级 provider 解析密钥与参数云函数场景下以环境变量大写命名替代secrets.toml依赖管理requirements.txt中声明dlt与dlt[bigquery]前者为 dlt 核心后者额外安装 BigQuery 目标端所需的依赖如google-cloud-bigquery。7. 常见问题与排查建议部署时报权限错误确认服务账号已具备Cloud Functions Developer角色使用 Secret Manager 时额外确认Secret Manager Secret Accessor角色函数超时大数据量上传时在部署命令或控制台将--timeout从 300 秒调高HTTP 函数上限 60 分钟密钥读取失败检查环境变量名是否与secrets.toml中的小写键一一对应并转为大写如api_key→API_KEYWebhook 收不到数据确认平台侧配置的 Trigger URL 正确、函数允许未认证调用并查看 GCF 日志中是否出现Event received and processed successfully.。总结本文完整演示了 dlt 管道在 Google Cloud Functions 上的两种典型部署形态一是以dlt initmain.pygcloud functions deploy构建的可手动触发管道二是面向实时事件的 Webhook 摄入函数。两者的核心都是「dlt 管道 HTTP 触发器」区别仅在于入口函数的编写方式与触发方式GET 手动触发 vs POST 事件推送。借助 dlt 统一的pipeline.run()接口与 BigQuery 目标端数据从 Notion 数据库或各类平台事件到数仓的链路可在数分钟内完成上云。【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考