本文内容主要给出基于PySpark程序，整合Spark Streaming和Kafka，实现实时消费和处理topic消息，为PySpark开发大数据实时计算项目提供基本参考。（未来将陆续更新基于Scala开发大数据实时计算项目的文章）
1 程序环境准备：
这里不再使用Spark的集群环境，因涉及的计算资源测试环境受限，目前两台虚拟机：1个vcore+2G内存，其中一台虚拟机启动Spark Streaming服务进程，另外一台虚拟机启动kafka进程。
虚拟机A：启动单实例kafka服务
虚拟机B：运行PySpark程序
在VM A，程序环境要求安装jdk1.8以上以及与kafka匹配版本的scala版本
版本兼容说明：
kafka：kafka_2.11-2.4.0 java：java version "1.8.0_11" scala： Scala 2.12.0这里需要注意：如果使用kafka_2.12版本以上，需要使用jdk1.8.0_212以上；kafka_2.12与jdk1.8.0_11有不兼容地方，kafka启动报错提示java.lang.VerifyError: Uninitialized object exists on backward branch 209。
1.1 基本配置
（1）配置单机zk这里无需依赖ZooKeeper集群，只需使用kafka自带的zk服务即可
vim /opt/kafka_2.11-2.4.0/config/zookeeper.properties
dataDir=/opt/zookeeper # zk的snapshot数据存储路径 clientPort=2181 # 按默认端口（2）配置kafka的，路径/opt/kafka_2.11-2.4.0/config/ server.properties
log.dirs=/opt/kafka-logs # 存放kafka数据目录 zookeeper.connect=127.0.0.1:2181 # 按默认连接本机zk即可1.2 启动zk和kafka
[root@nn kafka_2.11-2.4.0]# pwd/opt/kafka_2.12-2.4.0[root@nn kafka_2.11-2.4.0]# nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties 2>&1 &kafka server后台启动：
[root@nn kafka_2.11-2.4.0]# nohup bin/kafka-server-start.sh config/server.properties 2>&1 &1.3 测试单实例Kafka
对于kafka单节点而言，这里只能使用1个分区且1个replication-factor，topic名称为sparkapp
[root@nn kafka_2.11-2.4.0]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic sparkappCreated topic sparkapp.打开一个新的shell,用于启动producer
[root@nn kafka_2.11-2.4.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic sparkapp再打开一个新的shell,用于启动consumer
[root@nn kafka_2.11-2.4.0]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sparkapp在producer shell输入字符串，consumer端可以看到相应输出，说明单机的kafka可以正常运行，下面将使用Spark Streaming实时读取kafka的输入流
2 整合streaming和kafka
2.1 配置依赖包
具体说明参考官方文档spark streaming连接kafka需要依赖两个jar包（注意版本号）：
spark-streaming-kafka-0-8-assembly_2.11-2.4.3.jar： 下载链接
spark-streaming-kafka-0-8_2.11-2.4.4.jar： 下载链接
将这两个jar包放在spark 的jars目录下，需要注意的是：这两个jar包缺一不可，如果是在Spark集群上做测试，那么每个Spark节点都需要放置这两个jars包：
