针对 Kafka 开发者的 Azure 事件中心大规模流式传输（第 3 部分）：分析和可视化数据

在上一篇文章中，我们创建了一个 Azure Event Hub，允许基于 Kafka 的应用程序发送消息，并展示了如何通过对 Kafka 应用程序进行少量配置更改来连接到 Event Hubs。然后，我们创建了一个 Azure Function，它侦听 Event Hub 并将消息内容推送到 Cosmos DB。

在本文中，我们将进一步扩展，探索进入 Cosmos DB 的数据，创建更多 Function 来将同一条消息拆分成不同的数据类型，然后在 Power BI 中构建可视化以用于报告目的。

检查数据

让我们开始探索 Cosmos DB 实例中的当前数据。打开 Azure 门户，导航到我们在第二篇文章中创建的 Azure Cosmos DB 帐户。左侧应该有一个“Data Explorer”（数据浏览器）选项，它允许我们查看 Cosmos DB 中的所有数据。展开“Data”（数据）选项，应该有一个名为 `airlineData` 的数据库和一个名为 Customer（客户）的容器。如果我们点击容器下的 Items（项目）选项，Customer 中存储的所有项目都应该出现。当我们点击一个 id 时，该文档的架构及其包含的数据将显示在查看器的右侧。

创建更多 Function

从我们的数据中可以看到，JSON 对象包含一些嵌套元素。在构建报表时，最好将数据扁平化为单独的文档和表。为了实现这一点，让我们通过遵循以下步骤将当前 Function 分成四个不同的 Function：

1. 使用名称作为 ID 创建客户对象
2. 创建一个唯一的航班对象，不包含出发地或目的地子对象，并将航空公司的名称、停靠次数和价格与 ID 一起包含在内
3. 创建作为航班一部分的唯一出发地对象，包含出发地详细信息和航班 ID
4. 创建作为航班一部分的唯一目的地对象，包含目的地详细信息和航班 ID

让我们从修改当前的 `airline-customer-save` 开始，打开 `function.json` 文件，并将 `id` 替换为 `customer_name`，将 `partitionKey` 替换为 `/customer_name`。这会将客户表设置为使用客户的姓名作为标识符。我们还必须使用以下代码更新我们的 Function，以手动提取客户元素：

       event_body = json.loads(event.get_body().decode('utf-8'))
        customer = { 
            "customer_name": event_body["customer_name"], 
            "phoneNumber": event_body["phoneNumber"],
            "city": event_body["city"],
            "country": event_body["country"],
        }
        doc.set(func.Document.from_json(json.dumps(customer)))

这基本上会将我们保存的对象设置为仅包含我们感兴趣的元素。

更新此 Function 后，我们必须创建其他 Function 来以不同的方式处理消息。我们可以通过在 **Azure** 选项卡上使用“Create New Function”（创建新 Function）选项，或者通过复制 Function 文件夹并重命名组件来完成此操作。但是，在我们能够运行此程序之前，我们应该从头开始，通过删除 Cosmos DB 中的数据库来重新开始。我们可以通过重新打开 Cosmos DB 服务中的数据浏览器，右键单击数据库，然后选择“Delete Database”（删除数据库）来完成此操作。

删除数据库后，我们现在可以部署我们的 Function 并重新运行测试应用程序以生成更多消息。此示例生成了多条消息——大约 200 条，等待时间为 50 毫秒——以改进 Power BI 中的可视化效果。完成后，我们在 Cosmos DB 中拥有一个扁平化的数据集，我们可以在其上构建一些可视化。

连接 Power BI

Power BI 有两种版本——基于服务器的版本和基于桌面的版本——只有一些细微差别。本文使用的是来自Microsoft 网站的免费、基于桌面的版本。

下载并安装 Power BI，打开应用程序，然后在启动屏幕上点击 Get data（获取数据）。Power BI 可以与多个不同的数据库、对象和基于云的服务进行交互。我们特别寻找 Azure Cosmos DB。选择此选项，然后点击 connect（连接）。输入 Cosmos DB 概述页面的 URL。然后，输入数据库名称，然后点击 OK（确定）。

Power BI 然后尝试连接到 Cosmos DB，但在成功连接之前，我们必须输入密钥进行身份验证。您可以在 Cosmos DB 帐户的“Keys”（密钥）部分找到您的密钥。

如果您想了解更多有关 Azure Cosmos DB 的信息，或者分享您所知道的，请查看 Azure Cosmos DB Conf；这是一个与 Azure Cosmos DB 社区合作组织的免费在线虚拟开发者活动。

转换我们的数据

在通过 Cosmos DB 身份验证后，Power BI 应该会显示四个不同的表供我们选择。当我们点击这些表时，我们会看到每个数据结构都只有一个 Document（文档）列，其值为 Record（记录）。由于 Cosmos DB 是一个 NoSQL 数据库，每个文档都可以有自己的架构。因此，我们必须使用 Power BI 通过“Transform Data”（转换数据）选项来扁平化结构。

我们需要通过点击 Document 旁边的图标来展开我们每个表的 Document 列。这将显示一个屏幕，其中包含我们要展开的列的名称和要包含的表。对于每个表，选择适当的列，然后点击 OK（确定）。如果您保留原始列名，它会将 `Document` 添加到每个列名中。

我们还想对我们的数据进行一些其他更改，包括列类型和特定值。

打开 `flight` 表，在 `Document.price` 或 `price` 左侧的图标上，将 Column Type（列类型）更改为 Currency fixed decimal number（货币固定十进制数）。我们还想将 `Document.stops` 列中的 `non-stop` 值更改为 `0`。要做到这一点，选择列，点击“Transform”（转换）选项卡，然后选择“Replace Values”（替换值）。这将允许我们将 `non-stop` 值替换为 `0`。最后，将 `Document.stops` 的列类型更改为包含整数。

更新完这些值后，我们点击“Home”（主页）菜单上的 Close and Apply（关闭并应用）来将所有更改应用于我们的数据集。

构建可视化

我们可以构建的最直观的可视化之一是汇总表。主屏幕的右侧有一个 Visualization（可视化）选项卡。找到表格项并选择它，将其添加到我们的 Report（报表）窗格中。选中表格后，我们可以开始添加列。Power BI 会解释我们的数据，为我们创建基于表格的报表。如果我们从 `flight` 表的 Fields（字段）部分添加 `Airline`（航空公司）字段，我们可以看到所有不同的航空公司。此外，如果我们选择 `price`（价格）字段，表格将对航空公司价格进行求和并提供总计。我们甚至可以按不同的列标题排序。

我们也可以即时更改可视化，将表格变成更具视觉效果的内容。如果我们点击 Visualizations（可视化）选项卡上的堆叠条形图图标，表格就会变成一个排序的条形图。

Power BI 具有多种开箱即用的可视化效果，适用于任何数据集。例如，如果我们通过点击汉堡菜单图标并选择 Remove（移除）来删除此可视化，然后在画布上添加地图可视化，并将目的地国家/地区拖到 Location（位置）过滤器，我们就能看到人们飞往的所有目的地的地图。

我们只是触及了 Power BI 的能力。我们可以用许多其他方式来操作我们的数据，例如链接表、创建多值可视化和合并列。

摘要

在本系列中，我们构建了一个示例 Kafka 应用程序，并体验了少量配置更改如何使其能够连接到无服务器的 Event Hub 消息管道。我们创建了无服务器函数来处理这些消息，形成存储在 Cosmos DB 中的数据集。最后，我们使用该数据集来探索 Power BI 的潜在可视化效果。我们所有这些都是在不安装或管理复杂服务器基础设施的情况下完成的——并且没有离开 Python。

Microsoft 提供了很好的 视频资源来学习 Power BI，可以向您展示更多可能性。而且，通过 注册 Azure 时免费获得的 200 美元信用额度，您可以快速轻松地构建类似这样的东西。

要了解如何加速基础应用程序开发，请查看开发人员的低代码应用程序开发 7 步指南。