Azure Databricks大数据构建营 小试牛刀,大数据基础组件-ESG跨境

Azure Databricks大数据构建营 小试牛刀,大数据基础组件

Azure Databricks大数据构建营 小试牛刀

看过美剧《绝命毒师》（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:azureeventhubsspark库文件，以便安装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的价值。敬请期待！

特别声明：以上文章内容仅代表作者本人观点，不代表ESG跨境电商观点或立场。如有关于作品内容、版权或其它问题请于作品发表后的30日内与ESG跨境电商联系。