如何使用 Pipedream 和 InfluxDB 创建低代码工作流自动化

作者：Charles Mahler / 产品, 用例, 开发者
2022 年 2 月 16 日

导航至

现代软件开发的一个重要部分涉及使用 API。虽然使用第三方服务可以加速开发，但移动数据和将各种事物连接在一起可能非常枯燥。幸运的是，越来越多的工具可以帮助处理这些枯燥的事情，让您可以专注于更有趣的事情。Pipedream 就是其中一种工具。

在本教程中，您将学习如何连接 InfluxDB 和 Pipedream，以创建一个灵活的警报和通知系统，该系统可以通过 Pipedream 提供的 100 多种不同 API 的功能进行扩展。

什么是 Pipedream？

Pipedream 是一个集成平台，旨在帮助开发人员构建事件驱动型工作流。简单来说，Pipedream 通过提供与 400 多个不同 API 提供商的集成，使将大量不同工具组合在一起变得容易。

您可以将 Pipedream 视为编写您自己的自定义 API 集成与使用 Zapier 或 IFTTT 等低代码或无代码平台之间的中间地带。对于许多任务，您可以通过使用 Pipedream 提供的构建块来显着加快您的开发时间，但如果需要，您也可以选择编写自己的自定义代码。Pipedream 还提供了许多功能，通过其 UI 使测试和调试您的工作流更加容易。

Pipedream 的一些常见用例是

  • 将 SaaS 应用连接在一起
  • 自定义通知和警报
  • 数据库自动化
  • 移动应用或 JAMstack 站点后端
  • 速率限制

Pipedream 关键概念

使用 Pipedream 时，主要使用三个组件

  • 工作流 - 工作流是“容器”，包含您希望在处理 Pipedream 时对任何输入执行的所有操作。
  • 触发器 - 触发器定义工作流如何启动。它们可以是基本的 HTTP 请求、预定的时间或由第三方应用触发。
  • 步骤 - 步骤是工作流的构建块。它们在工作流被触发后按顺序运行。它们可以是 Pipedream 提供的预构建组件和操作，也可以是您自己的自定义代码。数据可以在每个阶段传递到下一步。

设置 InfluxDB

要遵循本教程，您将需要一个正在运行的 InfluxDB 实例和一个免费的 Pipedream 帐户。一旦您的 InfluxDB 实例运行起来，您需要创建一个 Bucket，该 Bucket 将用于测试并将警报发送到 Pipedream 以进行进一步操作。

创建 Bucket 后，您需要通过转到数据浏览器并单击脚本编辑器按钮来插入一些 示例数据

Inserting sample data by going to the Data Explorer and clicking the Script Editor button

在编辑器中输入以下 Flux 查询，并将 bucket 和 org 参数值更改为您刚刚创建的 Bucket 和您的帐户组织名称，然后单击提交。

import "experimental/csv"

relativeToNow = (tables=<-) =>
  tables
    |> elapsed()
    |> sort(columns: ["_time"], desc: true)
    |> cumulativeSum(columns: ["elapsed"])
    |> map(fn: (r) => ({ r with _time: time(v: int(v: now()) - (r.elapsed * 1000000000))}))

csv.from(url: "https://influx-testdata.s3.amazonaws.com/noaa.csv")
  |> relativeToNow()
  |> to(bucket: "your-bucket-name", org: "org-name")

Flux query - InfluxDB Script Editor - InfluxDB

此查询会将 NOAA 水样数据集存储在您的 Bucket 中，并修改时间戳，使其设置为相对于当前时间，这将使查询数据更容易。

设置 Pipedream

在您的 Pipedream 仪表板中，单击工作流选项卡，然后单击蓝色的“新建”按钮以添加新的工作流。您应该在屏幕上看到以下内容

Pipedream setup - dashboard

在本教程中，我们将使用 HTTP API 触发器，它为您提供一个 URL，每当 HTTP 请求发送到该端点时，都会触发工作流。单击 HTTP API 触发器按钮后，您的工作流应如下所示

Pipedream workflow - HTTP API

接下来，您可以添加一个 NodeJS 步骤。在本例中，我们将保持简单，只使用 console.log 来查看从 HTTP 触发器传递过来的事件对象。当我们在下一阶段创建警报时，这将向我们展示由 InfluxDB 警报创建并发送到 Pipedream 的对象。

Object created by the InfluxDB alert that is being sent to Pipedream.

最后，我们将发送一封电子邮件来通知自己何时发生错误。

Pipedream - alert email to notify about error

使用 Pipedream 的最佳好处之一可以在这里看到——UI 自动解析正在传递的事件，并允许您单击并从事件对象中抓取您想要的属性。您还可以重放过去的事件，尝试调试或修改您的工作流，以查看结果如何变化，这使得开发更加顺畅。

一旦您将通知电子邮件格式化为您想要的方式，请部署并保存您的工作流。请务必复制 HTTP API 触发器的 URL，因为我们将在下一步中需要它。

创建 InfluxDB 警报和检查

现在 Pipedream 已设置并准备好接收请求，您可以继续下一步，即在 InfluxDB 中创建警报。转到警报选项卡，然后单击按钮以创建通知端点。选择 HTTP 作为目标，命名您的端点，然后粘贴从您的 Pipedream URL 触发器复制的 URL，并选择 POST 作为 HTTP 方法。

InfluxDB - create notification endpoint

现在转到检查选项卡并创建一个 阈值 检查。这将打开一个数据浏览器视图，您可以在其中配置要检查的 Bucket 以及要查询的字段。选择您之前存储数据的 Bucket 以及您要发送警报的字段和位置。

InfluxDB - configure your check

在下一个选项卡中，您可以进一步配置您的检查。以下是一些更重要的注意事项

  • 每隔多久计划 - 这将计划检查运行的频率。默认情况下，检查每分钟运行一次
  • 标签 - 设置标签将允许您在稍后创建通知规则时筛选特定检查。
  • 状态消息 - 将与通知 HTTP 请求一起发送的消息，可以自定义以包含字段值、列名等。您可以在 文档中了解更多关于自定义此消息的信息。
  • 阈值 - 可以根据每个阈值级别的查询返回的值设置条件。在下一步中，您将看到如何使用阈值级别来筛选何时发送通知警报。

InfluxDB - further configure your check

对于本教程，您可以为您的查询和规则选择任何您想要的——只需确保您的阈值被触发，以便您获得一些发送到 Pipedream 的测试数据。

如果您计划在 Pipedream 中创建多个端点，您还需要向您的检查添加标签，以便您可以将每个检查筛选到正确的端点。

如果您想要更多地控制您的检查，您还可以选择直接查看由 UI 自动生成的 Flux 脚本。返回到检查选项卡，然后单击您刚刚创建的检查的设置图标。单击 编辑 ，您可以自定义 Flux 脚本。

InfluxDB checks - customize Flux script

您可以通过查看警报历史记录页面来确认您的检查是否正常运行。

InfluxDB - Alerts History

您需要做的最后一件事是创建通知规则。设置条件以匹配您要监视的标签和阈值级别，然后选择您之前创建的通知端点。如果您不创建此规则，您的检查将运行并存储在 _monitoring Bucket 中，但不会发生任何事情。

InfluxDB - Edit notification rule

结果

如果一切设置正确，那么您应该开始在您的 Pipedream 工作流中看到事件出现

Events - Pipedream workflow

如果我检查我的电子邮件，我也可以看到消息正确地传入

Email - Pipedream workflow

如果您不打算长期使用这些警报，请务必在 InfluxDB UI 中禁用它们，这样它们就不会在后台不断运行，并可能增加费用或耗尽您的免费套餐限制。

其他用例

现在您已经设置了基本的工作流以确认一切正常，您可以通过 Pipedream 访问的众多集成，获得大量选项来将事情提升到新的水平。在本节中，我将介绍一些关于 Pipedream 更高级用法的潜在想法。

扩展您当前的工作流

现在，警报工作流只会向您用于创建 Pipedream 帐户的地址发送电子邮件。如果您在生产环境中使用这种类型的警报，您将需要确保通过 API 调用抓取当前值班人员的联系信息，从而将警报发送给他们。

以下是一些可能对您有用的集成，可以查看一下

  • Twilio - 向您的值班工程师发送短信。
  • Slack - 如果您有用于支持或 SRE 的频道，那么在那里发送消息以通知他们警报可能是有意义的。
  • Zoom - Pipedream 可用于自动创建 Zoom 会议并发送事件响应链接。
  • 数据库集成 - Pipedream 与所有主要的数据库（如 Postgres、MySQL 和 MongoDB）都有集成。数据可以被查询并用作您工作流的一部分。附加数据可以帮助为必须处理警报的人员提供上下文。

应该注意的是，Flux 还通过其 标准库 包提供了与许多这些服务的集成，因此如果您不想要/不需要在 Pipedream 的工作流中使用多个步骤，那么直接通过 Flux 与 API 通信可能会更容易。

条件工作流

Pipedream 目前在用户界面中没有直接支持处理分支或条件工作流。目前的解决方案是创建一个由您的主工作流触发的新工作流。

您执行此操作的方式是在主工作流中使用 JavaScript 步骤，然后读取传入的数据并使用条件语句来确定接下来应该运行哪个工作流。您可以将当前工作流中的数据发送到触发下一个工作流的 HTTP 请求的正文中。根据您的用例，您可以为不同的条件创建多个不同的工作流。

丰富数据

在本教程的主要部分，InfluxDB 通过发送警报充当工作流的来源和触发器。InfluxDB 也可以用作工作流的中间步骤。一个例子是从外部来源收集和转换数据，然后再将其存储在 InfluxDB 中。

一个实际的例子是使用 Pipedream 的 Twitter 搜索触发器 步骤来提取您感兴趣的任何查询的推文。让我们以您公司的名称为例——您可以将这些推文存储在 InfluxDB 中，以跟踪随时间推移的趋势，并根据过去一小时内的推文数量设置警报，并在推文数量与历史数字相比突然增加时，通过另一个 Pipedream 工作流通知您的营销团队。这可以使用 Flux 提供的任何 内置分析函数 来完成，这些函数适合您的用例。

您还可以使用 Pipedream 创建一个中间步骤来 检查推文的情绪，以确定它是积极的还是消极的，并相应地自定义工作流，并将附加数据与 InfluxDB 一起存储，以便进行更准确的分析。

计划任务

Pipedream 还为希望定期运行的工作流提供了计划任务触发器。这可以用于抓取网站、处理数据，然后根据该数据触发其他操作。

下一步

从本文和使用 Pipedream 中获得的最大收获应该是，您真正受到的限制只是您的想象力。轻松访问数百种不同的工具和服务，为您提供了几乎无限种组合它们的方式。希望您可以将您在本文中学到的知识应用起来，并用它创造出一些令人惊叹的东西。我将为您提供一些其他资源的链接，如果您想更深入地了解本文中提到的一些主题，请查看这些资源。