Nutanix 数据管道和面向开发者的 AI

物联网边缘计算的核心在于实时分析以及理解用户如何与应用程序交互。在大数据物联网场景中，任何能够改进抽取、转换、加载 (ETL) 过程中的转换阶段的操作都将非常有益，因为您的边缘数据源可能多达数百甚至数千个。每个边缘设备都在向您的服务推送新的消息和数据负载，并期望高吞吐量和低延迟。

文章《面向开发者的 Nutanix Xi 物联网应用程序容器》讨论了 Xi IoT 如何允许您使用容器化应用程序来为您的边缘计算解决方案配置和部署自定义服务和运行时环境。

在本文中，我将更进一步，完全抽象化容器组件，并向您演示我们用于真实 AI 实现的原生数据管道。数据管道可以极大地简化将应用程序功能连接到数据源的过程，让您可以专注于开发转换逻辑，而不是构建运行时。

Xi IoT 上的数据管道

Xi IoT 平台上的数据管道支持完整的端到端 ETL 解决方案。数据管道是由输入流、转换函数和输出端点组成的一个集合。管道使您能够连接到数据源，无论其消息数据类型、物理位置、消息发送速度等如何。

要在 Xi IoT 管理控制台中查看已配置的数据管道，请点击“应用程序和数据”>“数据管道”。“列表”选项卡显示当前管道的列表。“可视化”选项卡提供卡片视图，显示数据源、管道和外部云环境之间的连接。“警报”选项卡显示与已配置管道相关的任何当前连接警报。

当您创建新的数据管道时，会提供一个可视化编辑器，您可以使用它将输入、转换和输出元素相互连接并创建管道。

数据管道以顺序方式工作。它们捕获来自数据源的输入，并将负载传递给转换函数。转换函数只需要专注于转换逻辑或分析，然后可以将它们的输出传递给其他转换函数、数据管道、存储或其他服务。

传感器和其他设备等数据源在 Xi IoT 服务的“基础设施”部分进行配置，然后作为数据源和数据流暴露给管道。对于数据管道，数据源或数据流在编辑器的“输入”区域中选择。

用于管道摄取的特定于数据源的所有配置都在设置数据源本身时完成，而不是在管道中完成。这意味着每个数据源仅配置一次，即可被多个管道使用。例如，您可能已配置

• 数据源类型，例如传感器或网关。
• 用于捕获消息的协议。
• 身份验证模式（如果有）。
• 源的 IP 地址。

配置 Xi IoT 中的数据源和边缘的一个关键功能是定义属性和类别，这些属性和类别有助于您在创建数据管道时查询和选择它们。在“基础设施”>“类别”下，您可以创建自定义类别并为数据源分配属性值。它可以是您环境中任何有意义的内容：传感器类型或用途的描述、数据协议、位置或任何允许轻松配置解决方案的内容。这些属性才是用作管道的输入标准，而不是每个单独的传感器。这意味着它们是动态的，并且非常容易扩展。

开发转换函数

作为开发人员，您的主要工作是编写在管道内部处理传入原始数据的函数。函数类似于您可能已经熟悉的无服务器计算函数，但可以更强大。在 Xi IoT 中，您可以直接在“应用程序和数据”>“函数”页面将转换函数编写（或上传）到服务中，从而创建可重用函数的库。Xi IoT 支持多种语言来编写函数，目前包括 Go、Node (JavaScript) 和 Python。

我们不仅提供这些语言的核心运行时，还提供与最广泛使用的平台和库集成的优化运行时。例如，我们有一个针对 TensorFlow 配置的运行时，您只需要编写利用 TensorFlow 的代码，而无需管理所有必需软件包的安装。

我们计划在未来支持更多运行时和语言。在本文撰写之时，我们已支持 TensorFlow，但您也可以利用外部 AI 服务——RESTful API，或 Python 或 Go 的软件包——来执行这些操作（更多内容将在未来的帖子中介绍）。

我们提供一个在线编辑器供您编写脚本，但如果您有本地编写的函数（符合我们的函数标准和要求），则只需选择要上传的文件即可。

在开发函数时，请记住我们平台用于调用您的函数的签名。您的函数需要有一个接受两个参数（一个上下文和一个将由平台传递的消息）的 main 函数。例如，在 Python 中，它看起来像这样：

def main(ctx,msg):

上下文 (ctx) 参数用于在您处理完消息 (msg) 后将输出返回给管道。以下示例将帮助您轻松理解这些参数之间的关系。在下面的示例中，我们将计算 AI 检测到的图像中的对象数量。

import json
import logging

def main(ctx,msg):
    input_data = json.loads(msg)
    detections = input_data['detections']
    result = {}
    object_counts = {}
    for i in range(len(detections)):
        detection = detections[i]
        label = detection['label']
        if label in object_counts:
            object_counts[label] = object_counts[label] + 1
        else:
            object_counts[label] = 1
    result['image'] = input_data['data']
    result['metadata'] = object_counts 
    output = json.dumps(result)
    return ctx.send(output)

我们从 msg 参数中捕获数据负载，并在函数体内进行处理。这就是数据管道如何帮助您开发函数的地方，因为您只需要检查传递的参数，而无需检查源或原始数据类型。数据管道负责将负载转发给相应的转换函数。

完成消息处理后，我们使用上下文参数中的 send 函数将消息转发给下一个函数或中间件。这使得管道能够将负载从一个转换传递到另一个转换。

除了主函数签名本身，您还可以根据需要添加任意数量的辅助函数。您不仅可以在函数内部添加子函数，还可以将一个函数链接到另一个函数。正如我们在示例项目中看到的，我们在管道中使用了多个函数，将图像从原始内容转换为 JPEG，处理图像以查找对象，然后在检测到的区域周围绘制边界框。

当然，我们也可以一步完成所有操作，但这会增加函数的整体复杂性及其处理消息的运行时。通过使用多个函数，我们可以让平台协调它们并为我们处理可伸缩性。此外，我们可以编写更多独立的专用函数，这些函数可以在其他管道中按需重用。

云端的输出数据

在我们的函数转换数据后，我们可以将数据输出到外部服务，以便用于其他目的——报告、下载、存储等。我们的平台支持数据输出到

• 云服务
• 您自己的数据存储库
• 其他数据管道
• 在容器化应用程序中运行的服务

在连接云实例时，我们目前支持使用标准 API 直接连接到多个 Amazon Web Services 和 Google Cloud Platform 服务。

在示例项目中，我们的输出被发送到边缘本身的一个服务。

您可以在 Xi IoT 的“基础设施”>“云配置文件”下配置输出服务。配置好云配置文件后，您可以使用“输出”窗格将内容提交到您的云存储。

如果使用 AWS，我们支持 S3、Kinesis 和 SQS 来摄取您云存储中的数据。这取决于您的选择以及业务用例，是选择边缘进行数据流处理，还是利用云存储来摄取处理过的数据。

结论

从本教程中可以看出，Xi IoT 平台的优势在于能够抽象掉您解决方案的一些底层细节，让您专注于连接数据管道中的重要元素。数据源和输出位置或服务可以一次性配置为基础设施，然后在构建数据转换管道时作为简单的输入和输出进行连接。转换本身也配置为可插拔、可重用的函数库。一旦设置好 Xi IoT 解决方案的构建块，构建或升级数据转换管道就如同连接这些组件一样简单。

除了这里演示的管理界面外，Xi IoT 还提供了一个 RESTful API，可用于以编程方式访问输入数据流、管道、函数和输出云配置文件。

所有这些功能都能让您能够轻松地配置、部署和管理基于物联网的实时分析应用程序。