使用Apache Spark构建实时分析Dashboard

2016 年 11 月 27 日

本篇博文中( Building Real-Time Analytics Dashboard Using Apache Spark ,作者: Abhinav Singh Sandeep Giri ),我们将学习如何使用 Apache Spark streaming,Kafka,Node.js,Socket.IO 和 Highcharts 构建实时分析 Dashboard。

问题描述

电子商务门户( http://www.aaaa.com )希望构建一个实时分析仪表盘,对每分钟发货的订单数量做到可视化,从而优化物流的效率。

解决方案

解决方案之前,先快速看看我们将使用的工具:
Apache Spark – 一个通用的大规模数据快速处理引擎。Spark 的批处理速度比 Hadoop MapReduce 快近 10 倍,而内存中的数据分析速度则快近 100 倍。更多关于Apache Spark 的信息。

Python – Python 是一种广泛使用的高级,通用,解释,动态编程语言。 更多关于Python 的信息。

Kafka – 一个高吞吐量,分布式消息发布订阅系统。 更多关于Kafka 的信息。

Node.js – 基于事件驱动的 I/O 服务器端 JavaScript 环境,运行在 V8 引擎上。 更多关于Node.js 的信息。

Socket.io – Socket.IO 是一个构建实时 Web 应用程序的 JavaScript 库。它支持 Web 客户端和服务器之间的实时、双向通信。 更多关于Socket.io 的信息。

Highcharts – 网页上交互式 JavaScript 图表。 更多关于Highcharts 的信息。

CloudxLab – 提供一个真实的基于云的环境,用于练习和学习各种工具。你可以通过在线注册立即开始练习。

如何构建数据Pipeline?

下面是数据Pipeline 高层架构图

(点击放大图像)

数据 Pipeline

实时分析 Dashboard

让我们从数据 Pipeline 中的每个阶段的描述开始,并完成解决方案的构建。

阶段 1

当客户购买系统中的物品或订单管理系统中的订单状态变化时,相应的订单 ID 以及订单状态和时间将被推送到相应的 Kafka 主题中。

数据集

由于没有真实的在线电子商务门户网站,我们准备用 CSV 文件的数据集来模拟。让我们看看数据集:

复制代码
DateTime, OrderId, Status
2016-07-13 14:20:33,xxxxx-xxx,processing
2016-07-13 14:20:34,xxxxx-xxx,shipped
2016-07-13 14:20:35,xxxxx-xxx,delivered

数据集包含三列分别是:“DateTime”、“OrderId”和“Status”。数据集中的每一行表示特定时间时订单的状态。这里我们用“xxxxx-xxx”代表订单 ID。我们只对每分钟发货的订单数感兴趣,所以不需要实际的订单 ID。

可以从 CloudxLab GitHub 仓库克隆完整的解决方案的源代码和数据集。
数据集位于项目的spark-streaming/data/order_data文件夹中。

推送数据集到 Kafka

shell 脚本将从这些 CSV 文件中分别获取每一行并推送到 Kafka。推送完一个 CSV 文件到 Kafka 之后,需要等待 1 分钟再推送下一个 CSV 文件,这样可以模拟实时电子商务门户环境,这个环境中的订单状态是以不同的时间间隔更新的。在现实世界的情况下,当订单状态改变时,相应的订单详细信息会被推送到 Kafka。

运行我们的 shell 脚本将数据推送到 Kafka 主题中。登录到 CloudxLab Web 控制台并运行以下命令。

复制代码
# Clone the repository
git clone https://github.com/singhabhinav/cloudxlab.git
# Create the order-data topic in Kafka
export PATH=$PATH:/usr/hdp/current/kafka-broker/bin
kafka-topics.sh --create --zookeeper localhost:2181
--replication-factor 1 --partitions 1 --topic order-data
# Go to Kafka directory
cd cloudxlab/spark-streaming/kafka
# Run the Script for pushing data to Kafka topic
# ip-172-31-13-154.ec2.internal is the hostname of broker.
# Find list of brokers in Ambari (a.cloudxlab.com:8080).
# Use hostname of any one of the brokers
# order-data is the Kafka topic
/bin/bash put_order_data_in_topic.sh ../data/order_data/
ip-172-31-13-154.ec2.internal:6667 order-data
# Script will push CSV files one by one to Kafka topic
after every one minute interval
# Let the script run. Do not close the terminal

阶段 2

在第 1 阶段后,Kafka“order-data”主题中的每个消息都将如下所示

复制代码
2016-07-13 14:20:33,xxxxx-xxx,processing

阶段 3

Spark streaming 代码将在 60 秒的时间窗口中从“order-data”的 Kafka 主题获取数据并处理,这样就能在该 60 秒时间窗口中为每种状态的订单计数。处理后,每种状态订单的总计数被推送到“order-one-min-data”的 Kafka 主题中。

请在 Web 控制台中运行这些 Spark streaming 代码

复制代码
# Login to CloudxLab web console in the second tab
# Create order-one-min-data Kafka topic
export PATH=$PATH:/usr/hdp/current/kafka-broker/bin
kafka-topics.sh --create --zookeeper localhost:2181
--replication-factor 1 --partitions 1 --topic order-one-min-data
# Go to spark directory
cd cloudxlab/spark-streaming/spark
# Run the Spark Streaming code
spark-submit --jars spark-streaming-kafka-assembly_2.10-1.6.0.jar
spark_streaming_order_status.py localhost:2181 order-data
# Let the script run. Do not close the terminal

阶段 4

在这个阶段,Kafka 主题“order-one-min-data”中的每个消息都将类似于以下 JSON 字符串

复制代码
{
"shipped": 657,
"processing": 987,
"delivered": 1024
}

阶段 5

运行 Node.js server

现在我们将运行一个 node.js 服务器来使用“order-one-min-data”Kafka 主题的消息,并将其推送到 Web 浏览器,这样就可以在 Web 浏览器中显示出每分钟发货的订单数量。

请在 Web 控制台中运行以下命令以启动 node.js 服务器

复制代码
# Login to CloudxLab web console in the third tab
# Go to node directory
cd cloudxlab/spark-streaming/node
# Install dependencies as specified in package.json
npm install
# Run the node server
node index.js
# Let the server run. Do not close the terminal

现在 node 服务器将运行在端口 3001 上。如果在启动 node 服务器时出现“EADDRINUSE”错误,请编辑 index.js 文件并将端口依次更改为 3002…3003…3004 等。请使用 3001-3010 范围内的任意可用端口来运行 node 服务器。

用浏览器访问

启动 node 服务器后,请转到 http://YOUR_WEB_CONSOLE:PORT_NUMBER 访问实时分析 Dashboard。如果您的 Web 控制台是 f.cloudxlab.com,并且 node 服务器正在端口 3002 上运行,请转到 http://f.cloudxlab.com:3002 访问 Dashboard。

当我们访问上面的 URL 时,socket.io-client 库被加载到浏览器,它会开启服务器和浏览器之间的双向通信信道。

阶段 6

一旦在 Kafka 的“order-one-min-data”主题中有新消息到达,node 进程就会消费它。消费的消息将通过 socket.io 发送给 Web 浏览器。

阶段 7

一旦 web 浏览器中的 socket.io-client 接收到一个新的“message”事件,事件中的数据将会被处理。如果接收的数据中的订单状态是“shipped”,它将会被添加到 HighCharts 坐标系上并显示在浏览器中。

截图

我们还录制了一个关于如何运行上述所有的命令并构建实时分析 Dashboard 的视频

我们已成功构建实时分析Dashboard。这是一个基本示例,演示如何集成Spark-streaming,Kafka,node.js 和socket.io 来构建实时分析Dashboard。现在,由于有了这些基础知识,我们就可以使用上述工具构建更复杂的系统。

希望本指南有帮助。 请随时留下您的意见。 在Twitter 上关注CloudxLab,获取有关新博客和视频的更新。

查看英文原文: Building Real-Time Analytics Dashboard Using Apache Spark


感谢杜小芳对本文的策划和审校。

给InfoQ 中文站投稿或者参与内容翻译工作,请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博( @InfoQ @丁晓昀),微信(微信号: InfoQChina )关注我们。

2016 年 11 月 27 日 16:566046

评论

发布
暂无评论
发现更多内容

安全&高可用架构

dony.zhang

高可用 安全 安全架构 高可用系统的架构 高可用系统的运维

LeetCode题解:20. 有效的括号,while循环replace,JavaScript,详细注释

Lee Chen

LeetCode 前端进阶训练营

对于三千万攀登者来说,云是安全绳,是登山杖,是趋顶之路

脑极体

《黑神话:悟空》出圈背后,国产3A游戏的技术新机

脑极体

奈学:红黑树(RedBlackTree)的概述

古月木易

架构师培训 -11 安全、高可用

刘敏

【Elasticsearch 技术分享】—— Elasticsearch 存储一条数据, put 过程是什么样子的?

程序员小航

Java elasticsearch 搜索 ES Lucene Elastic Search

30万奖金等你拿!Apache Flink 极客挑战赛入门指南(附Demo)

Apache Flink

flink

导致系统不可用的原因有哪些?

张磊

奈学:红黑树(RedBlackTree)的概述

奈学教育

AVL

第11周 安全和高可用

陆不得

架构师训练营第十一章作业

吴吴

Week11总结

张磊

图解 K8s 核心概念和术语

后端进阶

Docker Kubernetes 容器 云原生 k8s

区块链钱包系统开发服务商,区块链钱包应用搭建

13530558032

合约跟单APP开发模式,合约跟单系统源码搭建

13530558032

SpreadJS 纯前端表格控件应用案例:在线问卷系统

Geek_Willie

系统高可用

陈皮

Cause: java.sql.SQLTimeoutException: ORA-01013: user requested cancel of current

青乡之b

Druid

java 是什么?作用是什么?

InfoQ_34a83d636158

调度选得好,下班回家早

易观大数据

合同、封条、电梯……通通上链!

CECBC区块链专委会

区块链技术 监管平台

week11 作业

雪涛公子

架构师训练营第 11 周——练习

李伟

极客大学架构师训练营

极客大学架构师训练营---习题

李朋

XSKY S3 Console 为对象存储系统赋能

XSKY融合存储

朱嘉明:新冠肺炎疫情如何改变社会成本观念和结构

CECBC区块链专委会

社会结构 社会观念

区块链技术可提高数据可信性和安全性

CECBC区块链专委会

区块链技术 安全性

Java二进制和位运算,这一万字准能喂饱你

YourBatman

位运算 二进制

不可用与高可用

dongge

架构师课程第十一周总结

dongge

使用Apache Spark构建实时分析Dashboard-InfoQ