在Linux系统中打开浏览器，访问Spark官方下载地址，按照如下图下载。

Spark部署模式主要有四种：Local模式（单机模式）、Standalone模式（使用Spark自带的简单集群管理器）、YARN模式（使用YARN作为集群管理器）和Mesos模式（使用Mesos作为集群管理器）。 这里介绍Local模式（单机模式）的 Spark安装。我们选择Spark 3.3.2版本，并且假设当前使用用户名hadoop登录了Linux操作系统。

安装后，还需要修改Spark的配置文件spark-env.sh

编辑spark-env.sh文件(vim ./conf/spark-env.sh)，在第一行添加以下配置信息:

有了上面的配置信息以后，Spark就可以把数据存储到Hadoop分布式文件系统HDFS中，也可以从HDFS中读取数据。如果没有配置上面信息，Spark就只能读写本地数据，无法读写HDFS数据。 配置完成后就可以直接使用，不需要像Hadoop运行启动命令。 通过运行Spark自带的示例，验证Spark是否安装成功。

执行时会输出非常多的运行信息，输出结果不容易找到，可以通过 grep 命令进行过滤（命令中的 2>&1 可以将所有的信息都输出到 stdout 中，否则由于输出日志的性质，还是会输出到屏幕中）:

下面正式使用命令进入spark-shell环境，可以通过下面命令启动spark-shell环境：

该命令省略了参数，这时，系统默认是“bin/spark-shell --master local[*]”，也就是说，是采用本地模式运行，并且使用本地所有的CPU核心。

启动Spark

现在，你就可以在里面输入scala代码进行调试了。

比如，下面在命令提示符后面输入一个表达式“8 * 2 + 5”，然后回车，就会立即得到结果：

最后，可以使用命令“:quit”退出Spark Shell，如下所示：

使用winscp传输文件

接着在安装目录中使用下面命令创建一个Shell脚本文件，用于启动sbt：

该脚本文件中的代码如下：

保存后，还需要为该Shell脚本文件增加可执行权限：

然后，可以使用如下命令查看sbt版本信息：

1.安装maven ubuntu中没有自带安装maven，需要手动安装maven.我选择下载到本地在传输到虚拟机。 选择安装在/usr/local/maven中：

2.Java应用程序代码 在终端执行如下命令创建一个文件夹sparkapp2作为应用程序根目录

在 ./sparkapp2/src/main/java 下建立一个名为 SimpleApp.java 的文件（vim ./sparkapp2/src/main/java/SimpleApp.java），添加代码如下：

该程序依赖Spark Java API,因此我们需要通过Maven进行编译打包。在./sparkapp2中新建文件pom.xml(vim ./sparkapp2/pom.xml),添加内容如下，声明该独立应用程序的信息以及与Spark的依赖关系：

3.使用maven打包java程序 为了保证maven能够正常运行，先执行如下命令检查整个应用程序的文件结构:

文件结构如下图：

接着，我们可以通过如下代码将这整个应用程序打包成Jar(注意：电脑需要保持连接网络的状态，而且首次运行mvn package命令时，系统会自动从网络下载相关的依赖包，同样消耗几分钟的时间，后面再次运行mvn package命令，速度就会快很多):

如果运行上面命令后出现类似下面的信息，说明生成Jar包成功：

4.通过spark-submit 运行程序

可以通过spark-submit提交应用程序，该命令的格式如下：

最后，针对上面编译打包得到的应用程序，可以通过将生成的jar包通过spark-submit提交到Spark中运行，如下命令：

最后得到的结果如下:

winscp传输文件

执行如下步骤:

下面介绍Kafka相关概念,以便运行下面实例的同时，更好地理解Kafka.

Kafka集群包含一个或多个服务器，这种服务器被称为broker

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

Partition是物理上的概念，每个Topic包含一个或多个Partition.

负责发布消息到Kafka broker

消息消费者，向Kafka broker读取消息的客户端。

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

接下来在Ubuntu系统环境下测试简单的实例。Mac系统请自己按照安装的位置，切换到相应的指令。按顺序执行如下命令：

命令执行后不会返回Shell命令输入状态,zookeeper就会按照默认的配置文件启动服务,请千万不要关闭当前终端.启动新的终端，输入如下命令：

kafka服务端就启动了,请千万不要关闭当前终端。启动另外一个终端,输入如下命令

在创建主题时，

在较新版本（2.2 及更高版本）的 Kafka 不再需要 ZooKeeper 连接字符串，即- -zookeeper localhost:2181。使用 Kafka Broker的 --bootstrap-server localhost:9092来替代- -zookeeper localhost:2181。

topic是发布消息发布的category,以单节点的配置创建了一个叫dblab的topic.可以用list列出所有创建的topics,来查看刚才创建的主题是否存在。

可以在结果中查看到dblab这个topic存在。接下来用producer生产点数据：

并尝试输入如下信息：

然后再次开启新的终端或者直接按CTRL+C退出。然后使用consumer来接收数据,输入如下命令：

便可以看到刚才产生的三条信息。说明kafka安装成功。

Ubuntu 18.04.6 LTS自带Python3.6.9

Ubuntu系统中安装pip3，命令如下：

安装完pip3以后，可以使用如下Shell命令完成Flask和Flask-SocketIO这两个Python第三方库的安装以及与Kafka相关的Python库的安装：

这些安装好的库在我们的程序文件的开头可以直接用来引用。比如下面的例子。

from import 跟直接import的区别举个例子来说明。 import socket的话,要用socket.AF_INET,因为AF_INET这个值在socket的名称空间下。 from socket import* 是把socket下的所有名字引入当前名称空间。

使用winscp传输PyCharm 安装包

执行下面命令

然后，执行如下命令启动PyCharm：

执行上述命令之后，即可开启PyCharm。

​ 本案例采用的数据集压缩包为data_format.zip,该数据集压缩包是淘宝2015年双11前6个月(包含双11)的交易数据(交易数据有偏移，但是不影响实验的结果)，里面包含3个文件，分别是用户行为日志文件user_log.csv 、回头客训练集train.csv 、回头客测试集test.csv. 在这个案例中只是用user_log.csv这个文件，下面列出文件user_log.csv的数据格式定义：

用户行为日志user_log.csv，日志中的字段定义如下：

数据具体格式如下：

这个案例实时统计每秒中男女生购物人数，因此针对每条购物日志，我们只需要获取gender即可，然后发送给Kafka，接下来Spark Streaming再接收gender进行处理。

​ 本案例使用Python对数据进行预处理，并将处理后的数据直接通过Kafka生产者发送给Kafka，

接着可以写如下Python代码，文件名为producer.py：(具体的工程文件结构参照步骤一)

上述代码很简单，首先是先实例化一个Kafka生产者。然后读取用户日志文件，每次读取一行，接着每隔0.1秒发送给Kafka，这样1秒发送10条购物日志。这里发送给Kafka的topic为’sex’。

我们可以写一个KafkaConsumer测试数据是否投递成功，代码如下，文件名为consumer.py

在开启上述KafkaProducer和KafkaConsumer之前，需要先开启Kafka，命令如下：

启动zookeeper

启动Kafkaserver

在Kafka开启之后，即可开启KafkaProducer和KafkaConsumer。

在Pycharm启动producer.py和consumer.py

运行上面这条命令以后，这时，你会看到屏幕上会输出一行又一行的数字，类似下面的样子：

首先在/usr/local/spark/mycode新建项目目录

然后在kafka这个目录下创建一个kafka_test.py文件。

这个文件不做过多解释，因为这只是一个辅助文件，下面着重介绍工程主文件，文件名为KafkaTest.scala

上述代码注释已经也很清楚了，下面在简要说明下：

另外，需要注意，上面代码中有一行如下代码：

这行代码表示把检查点文件写入分布式文件系统HDFS，所以一定要事先启动Hadoop。如果没有启动Hadoop，则后面运行时会出现“拒绝连接”的错误提示。如果你还没有启动Hadoop，则可以现在在Ubuntu终端中，使用如下Shell命令启动Hadoop：

另外，如果不想把检查点写入HDFS，而是直接把检查点写入本地磁盘文件（这样就不用启动Hadoop），则可以对ssc.checkpoint()方法中的文件路径进行指定,比如下面这个例子：

编写好程序之后，下面介绍下如何打包运行程序。在/usr/local/spark/mycode/kafka目录下新建文件simple.sbt，输入如下内容：

然后，即可编译打包程序，输入如下命令

打包成功之后，接下来编写运行脚本，在/usr/local/spark/mycode/kafka目录下新建startup.sh文件，输入如下内容：

其中最后四个为输入参数，含义如下

运行脚本：

在同时开启Spark Streaming项目，KafkaProducer以及KafkaConsumer之后，可以在KafkaConsumer运行窗口

运行结果：

上篇文章说道Spark Streaming实时接收Kafka中topic为’sex’发送的日志数据，然后Spark Streaming进行实时处理，统计好每秒中男女生购物人数之后，将结果发送至Kafka，topic为’result’。在本章节，将介绍如何利用Flask-SocketIO将结果实时推送到浏览器。

Spark Streaming实时接收Kafka中topic为’sex’发送的日志数据，然后Spark Streaming进行实时处理，统计好每秒中男女生购物人数之后，将结果发送至Kafka，topic为’result’。在本章节，将介绍如何利用Flask-SocketIO将结果实时推送到浏览器。 文件目录结构如下：

首先我们创建如图中的app.py文件,app.py的功能就是作为一个简易的服务器，处理连接请求，以及处理从kafka接收的数据，并实时推送到浏览器。app.py的代码如下:

Python

这段代码实现比较简单，最重要就是background_thread函数，该函数从Kafka接收消息，并进行处理，获得男女生每秒钟人数，然后将结果通过函数socketio.emit实时推送至浏览器。

index.html文件负责获取数据并展示效果，该文件中的代码内容如下（也可以点击这里下载index.html文件）：

经过以上步骤，一切准备就绪，我们就可以启动程序来看看最后的效果。启动步骤如下：

1.确保kafka开启。 2.开启producer.py模拟数据流。 3.启动Spark Streaming实时处理数据。提示你可以在实时处理数据启动之后，把consumer.py的topic改成result，运行comsumer.py就可以看到数据处理后的输出结果。 4.启动app.py。如果你是使用pycharm客户端，那右键就可以了。当然也可以使用终端命令：

启动后的效果如下图：

pycharm运行图

说明：在本次实验中需要特别注意版本号，在运行代码、脚本、打包时，需要注意配置文件对应的jar包是否为对应环境版本

无法启动kafka 原因：新版的kafka启动命令已经修改，在较新版本（2.2 及更高版本）的 Kafka 不再需要 ZooKeeper 连接字符串，即- -zookeeper localhost:2181。使用 Kafka Broker的 --bootstrap-server localhost:9092来替代- -zookeeper localhost:2181。 无法使用sbt打包 原因：simple.sbt文件未修改为对应版本，sbt内容应于对应环境的spark、scala版本对应

Spark+Kafka构建实时分析Dashboard案例