本文共 3390 字，大约阅读时间需要 11 分钟。

看过美剧《绝命毒师》（Breaking Bad）的童鞋，恐怕都会对「老白」的化学知识和运用这些知识的能力留下深刻印象。毕竟，生活中一些常见的物品，通过对它们的成分进行一定组合往往能制造出很「不一般」的东西。

所以接下来你面对了一个新需求，老板要你实现这样的功能：一家连锁药店，希望通过抽样的方式检查止咳糖浆的销量，如果发现有人一次性购买10瓶以上就进行实时告警。你准备怎么做？

基于Azure Databricks的流计算就可以轻松实现。一起来看看吧。

本次《Azure   Databricks大数据构建营》系列文章，将主要围绕 Azure Databricks 以及其他配套服务，通过一系列实战教程告诉大家如何基于Azure云打造完全运行在云端的闭环大数据平台。

内容涵盖 ：Azure Databricks的基础知识，以及如何借助Azure   Databricks实现流计算（Structure   Streaming）、批处理（Spark   SQL）、安全访问控制以及机器学习能力等。

本系列的第一篇文章概括介绍了有关Azure Databricks的基础知识，错过的童鞋可以回看。而本篇，将带领大家小试牛刀，基于Azure   Databricks开发上文提到的那种实时告警系统。准备好，我们这就开始。

流计算架构说明

本篇涉及到的服务全都部署在中国区Azure上。根据模拟场景的描述，首先我们可以设计出如下所示的架构：

该解决方案将用到如下的产品和服务：

• Azure虚拟机：作为数据源，扮演Producer生产者，通过Python代码模拟客户购买行为，生成示例数据并通过SDK发送数据到Azure   EventHub。

• Azure EventHub：消息队列，做上下游生产者消费者服务的解耦，Entity ingestion负责接收Producer发送的数据，Entity alerting接收经过Databricks实时计算后的数据。

• Azure Databricks：订阅EventHub Entity   ingestion作为数据源，通过Structure   Streaming对数据进行实时处理后发送给Entity alerting。

• Azure LogicApp：订阅EventHub Entity   alerting并做邮件实时告警。

完整过程中的数据流是这样的：

1. Producer生产者发送数据

2. EventHub Entity ingestion（Ingestion 实体）

3. Databricks   Structured Streaming（流计算框架）

4. EventHub Entity alerting（Alerting 实体）

5. Logic App

Azure   Databrick Structure Streaming的实现

1. Terraform自动化部署

通过Terraform部署的服务组件包括Azure虚拟机、Azure   Databricks、EventHub、Logic App，具体的tf文件和变量可参见。每项服务Terraform   Azure Provider都由Resource支持，具体可参考。部署完成后的资源清单如下图所示，所有资源都部署在中国北二区域。

2.  Producer代码发布

模拟的生产者代码通过虚拟机发布，通过调用Azure EventHub的SDK将消息写入，具体代码可见，几个重要配置简单说明下：

• azure.eventhub：Azure EventHub SDK包，需要通过pip3 install   azure.eventhub来指定安装。

• create_batch()：通过该方法批量发送数据，本次示例以1条消息为1个批次发送到EventHub   Entity ingestion。

• CONNECTION_STR：Azure EventHub   Endpoint，该连接字符串可以在门户上Shared access policies的Connection string–primary key中查看。

• EVENTHUB_NAME：写入的EventHub Entity   Name。

3. Azure Databricks集群配置及Structure   Streaming Notebook的集成

Azure Databricks的创建过程是：首先在Azure上创建一个Databricks实体，然后在此基础上在实体内部创建Workspace以及Cluster，再提交Job。每个Databricks资源都有唯一的ID和Endpoint与之对应，以便能够进行Restful   API调用，集群通过Databricks门户创建即可。

本示例将创建1个Driver和2个Worker，共计3个节点的Standard   Cluster，Databricks版本为6.4（包含Apache Spark   2.4.5、Scala   2.11）。如果需要做机器学习相关计算，可启用集成GPU/ML框架的版本，详细说明见，这里不做赘述。

集群状态变为Running就意味着就绪可以使用了。不过在导入Python   Notebook之前，需要通过Maven   Central安装com.microsoft.azure:azure-eventhubs-spark库文件，以便安装Spark连接Azure EventHub   Connector，需要注意库文件的版本要匹配。

Notebook可以直接在门户里新建写入，也可以在VS   Code等IDE中编写完之后发布。本文采用第二种模式，原因是IDE丰富的插件可以提高效率。具体的Notebook 本文不做展示，放在，有需要的童鞋可以自行查看。通过import导入后，附上导入后的截图并做几点说明：

整个Notebook分为三个阶段：

• 第一阶段：从EventHub Entity   ingestion读取Producer写入的数据，通过Streaming   DataFrames 的spark.readStream   ()创建。

• 第二阶段：通过DataFrame丰富的函数做字段筛选，筛选出来我们需要的字段。

• 第三阶段：回写EventHub Entity   alerting，通过Streaming   DataFrames的spark.writeStream   ()流式写入，注意利用Checkpoint方便任务终止再运行。

当Producer运行起来后，EventHub就会不断有数据写入，所以能看到Spark的Input Records图像。对于每一个Job，都能看到对于该任务分配的资源和Spark参数配置项。

4. Logic APP配置邮件告警

经过Azure   Databricks的数据筛选后，筛选出来的Messages都写入了EventHub Entity alerting中，此时通过LogicApp来定义一个自动化的工作流来进行邮件告警。具体创建过程选择Blank，然后自己创建Step即可，当然Azure门户上的示例模板也可以用来参考，如下图所示：

第一步订阅EventHub   Entity alerting，第二步集成Outlook邮件接口发送告警邮件。所以当目标消息被筛选出来之后，LogicApp就按照定义的邮件内容（本文是消息内容和时间戳）来发送邮件，发送邮件的截图如下：

总结

总体上，一个通过消息队列Azure EventHub以及Databricks做流计算处理的示例就完成了。

如果消息生产者Producer不断产生消息，那么整个任务就会一直运行下去，当出现目标消息的时候就会不断的持续告警。在Spark推出Structure   Streaming后，也解决了Spark   Streaming micro batch的局限性。

本文开头提及的需求已经顺利实现！

有关如何借助Azure   Databricks实现流计算（Structure   Streaming）的内容就是这些了。随后的内容中，我们还将介绍如何实现批处理（Spark SQL）、安全访问控制和机器学习能力。通过这些真实场景中的最佳实践分享，也能帮助大家更清楚地认识到Azure Databricks的价值。敬请期待！

转载地址：http://unrtf.baihongyu.com/