如何将 AWS Lambda 无服务器函数与 InfluxDB 结合使用

作者：Charles Mahler / 产品, 用例, 开发者
2021 年 12 月 07 日

导航至

对于时间序列工作负载，无服务器函数能够向上和向下扩展的能力是一项主要优势，特别是对于像物联网设备这样的应用场景，它们可能具有间歇性连接，并且可能会突然以突发方式发送数据。在这种情况下，当您可以使用无服务器函数并且仅为您使用的计算资源付费时，为服务器 24/7 全天候运行付费是没有意义的。

在本文中，您将学习如何将无服务器函数与 InfluxDB 结合使用，同时了解 InfluxDB 平台提供的一些功能。

教程要求

要学习本教程，您需要一个可用的 InfluxDB 实例和一个 AWS 账户。 启动并运行 InfluxDB 账户最快、最简单的方法是创建一个免费的 InfluxDB Cloud 账户。 另一种选择是安装 InfluxDB 的开源版本。

我们的无服务器函数本教程使用 AWS Lambda，这需要一个 AWS 账户才能使用。 您也可以使用其他无服务器服务，例如 Google Cloud Functions 或 Azure Functions； 设置完成后，代码将在很大程度上以相同的方式工作。

使用 AWS Lambda 将数据写入 InfluxDB

您需要做的第一件事是将一些数据导入 InfluxDB。 有多种方法可以做到这一点

  • 使用 InfluxDB UI – InfluxDB 用户界面 允许您上传 CSV 或文本文件（以行协议格式格式化），或者直接键入行协议并通过 UI 提交。
  • Telegraf – Telegraf 是一个服务器代理，它具有超过 200 种不同工具和服务的插件，可用于收集数据并将其输出到 50 多个不同的目标位置。 其中一个输出是 HTTP 插件，它将数据作为 HTTP post 请求发送，可以将其定义为 AWS Lambda 函数
  • InfluxDB CLI – InfluxDB CLI 可用于将数据写入 InfluxDB 以及从命令行执行许多其他任务。
  • InfluxDB 客户端库 – InfluxDB 为 14 种不同的编程语言提供客户端库。

在本教程中，我们将使用 Javascript 客户端库 和 AWS Lambda 将数据写入 InfluxDB。 在此示例中，我们将通过使用 API Gateway 的 HTTP 请求来完成此操作。

在 AWS 控制台中，您需要创建一个 Lambda 函数。 完成后，您需要创建一个 Lambda Layer，这将允许您在 Lambda 函数中使用 InfluxDB 客户端库和任何其他 NPM 包。 为此，您需要在本地计算机上创建要使用的节点模块和 package.json 文件，然后将其作为 zip 文件上传。 以下是您希望文件夹结构的外观

Lambda functions - InfluxDB

这里需要注意的关键事项是外部目录**必须**命名为“nodejs”。 在该文件夹中，您只需要运行 npm init 并安装您的依赖项，就像对待任何正常的 nodeJS 项目一样。 跟随本教程所需的唯一包是 InfluxDB javascript 客户端库，可以使用以下命令安装

npm install --save @influxdata/influxdb-client

然后，您可以使用操作系统提供的任何 zip 压缩实用程序来压缩 nodejs 目录并将其作为层上传。 为此，您只需转到 AWS Lambda 仪表板，查看侧边栏上写着“附加资源”的位置，然后单击“层”选项卡。 进入后，您可以单击橙色的“创建层”按钮。 给它一个名称，上传 zip 文件，然后选择与您要用于 Lambda 函数的任何版本匹配的 NodeJS 版本运行时。

Nodejs - lambda function

创建此层的好处是您可以重复使用它，以便您可以在多个 Lambda 函数中访问 InfluxDB 客户端库，而无需再次经历此过程。 您还可以为函数使用多个层，以便您可以为您的用例“分层”不同的功能和环境。

完成设置后，您现在需要从主 Lamda 仪表板创建 Lambda 函数。 给它任何您想要的名称，选择 Node.js 14.x 运行时，然后点击“创建函数”按钮。

在特定于功能的仪表板中，您需要添加刚刚创建的层。 这可以通过滚动到仪表板底部并单击“添加层”来完成，然后在打开的页面上，您将单击“自定义层”选项卡并选择您之前创建的层。

对于触发器，您需要使用 Amazon API Gateway 触发器，以便您可以通过 HTTP 请求调用该函数。 选择“创建 API”并使其成为具有开放安全性的 HTTP API 类型。 创建 Amazon API Gateway 后，单击它并复制 API 端点 URL，这将稍后用于发出我们的 API 请求。

在为 Lambda 函数编写代码之前，您需要使用您的 InfluxDB 账户令牌和凭据进行身份验证。 要设置所有这些，您可以查看本教程，其中介绍了基础知识。 您将需要 4 件事来运行代码

  • 账户组织名称 – 这是在您创建账户时创建的，可以在您的设置中找到
  • 存储桶名称 – 您要存储数据的 InfluxDB 存储桶的名称
  • 令牌 – 用于验证您的请求的 API 令牌
  • InfluxDB 实例 URL – 如果您使用的是云账户，则此 URL 将取决于您在账户创建期间选择的云区域，此 URL 将是您的仪表板 URL 的子域。 您可以在此处找到云 URL 的完整列表。 如果您使用的是开源版本，它将在 localhost 上，可以在此处找到 URL。

在您的 Lambda 函数中，您需要进入“配置”选项卡并为每个值创建一些环境变量，以便可以将它们分配给变量。

InfluxDB ingest

完成所有设置后，让我们开始编写代码。 以下是 Lambda 函数的所有代码。 此模块会将数据写入我们的 InfluxDB 实例，并在内联注释中解释每个阶段发生的事情

//import InfluxDB client, this is possible thanks to the layer we created
const {InfluxDB, Point, } = require('@influxdata/influxdb-client')

//grab environment variables
const org = process.env.org
const bucket = process.env.bucket
const token = process.env.token;
const url = process.env.url

//lambda event handler, this code is ran on every external request
exports.handler =  async (event) => {
    //parse the expected JSON from the body of the POST request
    var body = JSON.parse(event.body)

    //create InfluxDB api client with URL and token, then create Write API for the specific org and bucket
    const writeApi = await new InfluxDB({url, token}).getWriteApi(org, bucket);

    //create a data point with health as the measurement name, a field value for heart beat, and userID tag
    const dataPoint = new Point('health')
        .tag('userID', body['userID'])
        .floatField('heartRate', body['heartbeatRate'])

    //write data point
    await writeApi.writePoint(dataPoint)

    //close write API
    await writeApi.close().then(() => {
        console.log('WRITE FINISHED')
    })

    //send back response to the client
    const response = {
        statusCode: 200,
        body: JSON.stringify('Write successful'),
    };
    return response;
};

要测试此函数并确保一切正常，您可以使用任何您想要的工具来发出 HTTP 请求。 我将使用 Postman。 如果您想了解有关如何使用 Postman 的更多信息，您可以查看本指南，其中涵盖了 Postman 的基础知识以及使用 InfluxDB REST API。 Postman 中的请求将如下所示

Authenticate - Lambda function

如果一切正常，您应该收到响应“写入成功”。 在使用不同的值和用户 ID 向函数发出一些请求后，您可以转到 InfluxDB UI 中的数据浏览器选项卡并查看您的数据。 单击您创建的存储桶名称，您应该会看到可用的“健康”指标。 您可以在此处看到两条不同的颜色线，因为我提交了带有两个不同用户 ID 标签的请求。 尝试在仪表板中玩转过滤选项和不同的图表样式，看看您可以做什么。

InfluxDB UI - Lambda function

处理存储在 InfluxDB 中的数据

现在您已经在 InfluxDB 中拥有了一些数据，让我们回顾一下您可以使用这些数据执行的一些示例。

使用 AWS Lambda 查询 InfluxDB

就像您可以使用 InfluxDB 客户端库将数据写入 InfluxDB 一样，您也可以使用它们来查询数据。 您可能想要这样做的一个原因是为了向最终用户显示自定义 UI。 有关此示例，您可以查看本教程，其中展示了如何使用 Nivo ReactJS 图表库。 通过这种方法，您可以使用 JavaScript 客户端库显示数据。

使用我们上面编写的示例“心跳数据”，您可以根据用户 ID 过滤数据，然后向他们显示他们的健康数据。 为此，您可以使用 AWS Lambda 函数为您的应用程序发出 API 请求。 虽然您可以使用 JavaScript 从浏览器查询 InfluxDB，但这会将您的 API 密钥暴露给公众，这是一个安全风险。 通过使用无服务器函数，您可以确保 API 密钥的安全，同时不必担心管理服务器。 您所要做的就是编写请求数据的代码。

客户端库的另一个用例可能是数据分析。 虽然 Flux 能够完成您在数据分析方面想要的几乎任何事情，但您可能更熟悉另一种编程语言，或者想要利用某些库和语言生态系统。 您可以查看本指南，了解如何使用 Python 从 InfluxDB 中提取数据，然后使用 Pandas 数据帧转换该数据。 一个更复杂的示例是基于此构建的股票市场示例，该示例展示了如何使用 Sci-kit learn 和 Keras 进行价格预测。

将 InfluxDB 警报和任务与 AWS Lambda 结合使用

存储在 InfluxDB 中的数据的其他常见用例与高级数据处理有关。 例如，您可以根据存储的数据包含某种数据转换和警报。 对于数据转换，InfluxDB 提供任务，它允许您在特定数据桶上运行计划的 Flux 脚本，然后将该数据输出到新的数据桶中。 一些常见的示例包括下采样数据，通过降低存储数据的精度来降低存储成本。 在另一个示例中，您可以通过连接来自另一个来源（例如关系数据库）的数据来丰富数据。 其他一些常见的用例包括清理数据、预测、创建报告、事件触发器和生成数据。

许多这些数据转换任务也可以使用无服务器框架和函数来完成。 您可以使用辅助服务来安排函数以定期间隔运行，但从成本和性能的角度来看，很难胜过在数据库内部使用 InfluxDB 任务来完成此类工作，并且从可用性的角度来看，连接所有内容是一件苦差事。

您还可以监控您的数据并在定义的阈值和参数不满足您定义的条件时创建警报。 您可以设置检查这些阈值的频率，然后通过添加端点来采取措施，以便在您的数据超出范围时发送警报。 InfluxDB 提供与 Slack 和 PagerDuty 的内置集成，用于在 UI 中发出警报，Flux 还具有许多其他可用的第三方集成，例如 Discord、Telegram、ServiceNow 等。 如果您想要最大的灵活性，您还可以选择定义 HTTP 端点。 您可以创建一个无服务器端点，然后使用该无服务器函数处理您的警报。 一个示例可能是创建 AWS Lambda 函数，该函数接收警报，然后使用 Twilio 或其他 SMS 服务向待命人员发送短信。

总结

这里真正要说明的是，通过使用 InfluxDB，您可以作为开发人员获得最大的灵活性。 您可以熟悉 Flux 并利用 InfluxDB 提供的所有内置功能，同时在需要时可以选择使用您喜欢的编程语言。

如果您想了解更多关于如何将 InfluxDB 与无服务器函数结合使用以及如何使用 InfluxData 平台 的其他优秀资源，您可以查看以下链接

常见问题解答

“无服务器”和 AWS Lambda 无服务器函数之间有什么区别？

区别在于 Lambda 是专门的 AWS 供应商服务，用于提供无服务器功能。

相比之下，“无服务器计算”仅指使用任何供应商系统将功能外包给远程服务器。

一些专家解释说，“无服务器”是通用术语，而 Lambda 是 AWS 特定的术语。

为什么 Lambda 是无服务器的？

Lambda 不是无服务器的，因为函数在完全不使用任何服务器的情况下运行。 AWS lambda 是无服务器的，因为客户端无需使用自己的服务器，而是可以在供应商的服务器上运行单个函数，因此从客户端的角度来看它是无服务器的。

如何使用无服务器部署 Lambda 函数？

简短的回答是，您将 AWS lambda 服务集成到您的应用程序和 API 中，以便在给定的无服务器框架内利用该无服务器功能。

AWS 无服务器函数有哪些好处？

无服务器函数是 AWS 提供的无服务器计算服务。 它是一个事件驱动的计算平台，允许您上传代码以构建无服务器应用程序，而无需管理基础设施。 在无服务器架构中，所有基础设施维护（例如操作系统配置、补丁和安全更新）都由 AWS 云提供商处理。 无服务器函数有以下几个好处

  • 成本节省： 无服务器函数具有成本效益，因为您只需在事件发生且代码运行时才需要为数据处理付费。
  • 安全性： 无服务器函数在容器中运行，这些容器将其运行时环境与其他无服务器应用程序和外部来源隔离。
  • 速度： 无服务器函数具有高性能，因为它们允许根据应用程序的流量自动向上或向下扩展。
  • 简易性： 团队可以使用 AWS 服务专注于应用程序设计、部署和交付，而无需担心基础设施管理、容量配置和硬件维护。
  • 可靠性： 无服务器函数不易发生故障。 无服务器应用程序是云中一系列互连的服务，使其自然而然地具有冗余性。
  • 可扩展性： 您可以使用函数即服务 (FaaS) 模型根据需要扩展单个进程，而不是增加整个无服务器应用程序的总资源分配。 这种灵活性降低了不必要的数据处理成本，并提高了整体应用效率。
AWS 基础设施中有哪些服务是无服务器的？
  • AWS Lambda 函数： AWS Lambda 函数是无服务器框架的一部分。 这些函数运行事件驱动的无服务器应用程序，其中计算资源由云提供商自动管理。
  • Amazon API Gateway： Amazon API Gateway 是无服务器框架中的一项 AWS 服务，允许开发人员构建、测试、部署和监控 API。
  • Amazon DynamoDB： Amazon DynamoDB 是无服务器堆栈的一部分，它是一项完全托管的 NoSQL 数据库服务，可让您卸载管理任务，例如设置、配置、扩展
  • Amazon S3： Amazon Simple Storage Service (S3) 是 Amazon Web Services (AWS) 存储平台，为对象提供 S3 存储。
  • Amazon Kinesis： Amazon Kinesis 是一项基于实时事件的流服务
  • Amazon Aurora： Amazon Aurora 是一项完全托管的 Amazon Relational Database Service (RDS)，可处理所有管理任务。 它提供精细的时间点恢复。
  • AWS Fargate： AWS Fargate 是一种无服务器计算引擎，它使用按需付费模式，其中计算时间仅在您使用该服务时才在您的计算实例上收费。
  • Amazon SNS： Amazon Simple Notification Service (SNS) 是一项简单、快速、可靠、高度可扩展的消息传递服务
AWS Lambda API 如何工作？为什么它与 AWS 资源一起使用？

AWS Lambda 是一项基于无服务器事件的服务，用于运行代码。 Lambda 应用程序自动为您管理计算资源。 它与 Fargate 的不同之处在于定价模式。 Lambda 按每次调用和每个事件的持续时间收费。 另一方面，Fargate 按 CPU 和内存每秒收费。

AWS Step Functions 是无服务器的吗？

是的。Step Functions 是无服务器的。它是一种编排服务，可让您创建函数，通过结合使用 AWS Lambda 和其他 AWS 服务，为 Web 应用程序构建无服务器工作流。无服务器工作流可帮助您监控工作流中的每个步骤，以验证其是否按预期顺序运行。