demo2 Kafka+Spark Streaming+Redis实时计算整合实践 foreachRDD输出到redis
基于Spark通用计算平台，可以很好地扩展各种计算类型的应用，尤其是Spark提供了内建的计算库支持，像Spark Streaming、Spark SQL、MLlib、GraphX，这些内建库都提供了高级抽象，可以用非常简洁的代码实现复杂的计算逻辑、这也得益于Scala编程语言的简洁性。这里，我们基于1.3.0版本的Spark搭建了计算平台，实现基于Spark Streaming的实时计算。
我们的应用场景是分析用户使用手机App的行为，描述如下所示：
- 手机客户端会收集用户的行为事件（我们以点击事件为例），将数据发送到数据服务器，我们假设这里直接进入到Kafka消息队列
- 后端的实时服务会从Kafka消费数据，将数据读出来并进行实时分析，这里选择Spark Streaming，因为Spark Streaming提供了与Kafka整合的内置支持
- 经过Spark Streaming实时计算程序分析，将结果写入Redis，可以实时获取用户的行为数据，并可以导出进行离线综合统计分析
Spark Streaming介绍
Spark Streaming提供了一个叫做DStream（Discretized Stream）的高级抽象，DStream表示一个持续不断输入的数据流，可以基于Kafka、TCP Socket、Flume等输入数据流创建。在内部，一个DStream实际上是由一个RDD序列组成的。Sparking Streaming是基于Spark平台的，也就继承了Spark平台的各种特性，如容错（Fault-tolerant）、可扩展（Scalable）、高吞吐（High-throughput）等。
在Spark Streaming中，每个DStream包含了一个时间间隔之内的数据项的集合，我们可以理解为指定时间间隔之内的一个batch，每一个batch就构成一个RDD数据集，所以DStream就是一个个batch的有序序列，时间是连续的，按照时间间隔将数据流分割成一个个离散的RDD数据集，如图所示（来自官网）：
我们都知道，Spark支持两种类型操作：Transformations和Actions。Transformation从一个已知的RDD数据集经过转换得到一个新的RDD数据集，这些Transformation操作包括map、filter、flatMap、union、join等，而且Transformation具有lazy的特性，调用这些操作并没有立刻执行对已知RDD数据集的计算操作，而是在调用了另一类型的Action操作才会真正地执行。Action执行，会真正地对RDD数据集进行操作，返回一个计算结果给Driver程序，或者没有返回结果，如将计算结果数据进行持久化，Action操作包括reduceByKey、count、foreach、collect等。关于Transformations和Actions更详细内容，可以查看官网文档。
同样、Spark Streaming提供了类似Spark的两种操作类型，分别为Transformations和Output操作，它们的操作对象是DStream，作用也和Spark类似：Transformation从一个已知的DStream经过转换得到一个新的DStream，而且Spark Streaming还额外增加了一类针对Window的操作，当然它也是Transformation，但是可以更灵活地控制DStream的大小（时间间隔大小、数据元素个数），例如window(windowLength, slideInterval)、countByWindow(windowLength, slideInterval)、reduceByWindow(func, windowLength, slideInterval)等。Spark Streaming的Output操作允许我们将DStream数据输出到一个外部的存储系统，如数据库或文件系统等，执行Output操作类似执行Spark的Action操作，使得该操作之前lazy的Transformation操作序列真正地执行。
Kafka+Spark Streaming+Redis编程实践
下面，我们根据上面提到的应用场景，来编程实现这个实时计算应用。
首先，创建一个scala工程，创建方法见 三、使用maven创建scala工程（scala和java混一起）
引入kafka、redis、json等相关的包，pom.xml 如下：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><parent><artifactId>fvp</artifactId><groupId>com.sf.fvp</groupId><version>0.0.1-SNAPSHOT</version></parent><groupId>com.sf</groupId><artifactId>scalademo3</artifactId><version>1.0-SNAPSHOT</version><name>${project.artifactId}</name><description>My wonderfull scala app</description><inceptionYear>2015</inceptionYear><licenses><license><name>My License</name><url>http://....</url><distribution>repo</distribution></license></licenses><properties><maven.compiler.source>1.6</maven.compiler.source><maven.compiler.target>1.6</maven.compiler.target><encoding>UTF-8</encoding><scala.version>2.11.5</scala.version><scala.compat.version>2.11</scala.compat.version></properties><dependencies><!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.10 --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.0.2</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><version>2.0.2</version></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>2.0.2</version></dependency><dependency><groupId>net.sf.json-lib</groupId><artifactId>json-lib</artifactId><version>2.2.3</version></dependency><dependency><groupId>org.codehaus.jettison</groupId><artifactId>jettison</artifactId><version>1.3.8</version></dependency><dependency><groupId>net.sf.ezmorph</groupId><artifactId>ezmorph</artifactId><version>1.0.6</version></dependency><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.5.2</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId><version>2.2</version></dependency><!-- Test --><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.11</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-core_${scala.compat.version}</artifactId><version>2.4.16</version><scope>test</scope></dependency><dependency><groupId>org.scalatest</groupId><artifactId>scalatest_${scala.compat.version}</artifactId><version>2.2.4</version><scope>test</scope></dependency></dependencies><build><sourceDirectory>src/main/scala</sourceDirectory><testSourceDirectory>src/test/scala</testSourceDirectory><plugins><plugin><!-- see http://davidb.github.com/scala-maven-plugin --><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.2.0</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals><configuration><args><arg>-make:transitive</arg><arg>-dependencyfile</arg><arg>${project.build.directory}/.scala_dependencies</arg></args></configuration></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.18.1</version><configuration><useFile>false</useFile><disableXmlReport>true</disableXmlReport><!-- If you have classpath issue like NoDefClassError,... --><!-- useManifestOnlyJar>false</useManifestOnlyJar --><includes><include>**/*Test.*</include><include>**/*Suite.*</include></includes></configuration></plugin></plugins></build>
</project>二、写了一个Kafka Producer模拟程序，用来模拟向Kafka实时写入用户行为的事件数据，数据是JSON格式，示例如下：
{"uid":"068b746ed4620d25e26055a9f804385f","event_time":"1430204612405","os_type":"Android","click_count":6}一个事件包含4个字段：
- uid：用户编号
- event_time：事件发生时间戳
- os_type：手机App操作系统类型
- click_count：点击次数
下面是我们实现的代码，如下所示：
package com.sf.scalademo3import java.util.Properties
import scala.util.Properties
import org.codehaus.jettison.json.JSONObject
import kafka.javaapi.producer.Producer
import kafka.producer.KeyedMessage
import kafka.producer.KeyedMessage
import kafka.producer.ProducerConfig
import scala.util.Randomobject KafkaEventProducer {private val users = Array("4A4D769EB9679C054DE81B973ED5D768", "8dfeb5aaafc027d89349ac9a20b3930f","011BBF43B89BFBF266C865DF0397AA71", "f2a8474bf7bd94f0aabbd4cdd2c06dcf","068b746ed4620d25e26055a9f804385f", "97edfc08311c70143401745a03a50706","d7f141563005d1b5d0d3dd30138f3f62", "c8ee90aade1671a21336c721512b817a","6b67c8c700427dee7552f81f3228c927", "a95f22eabc4fd4b580c011a3161a9d9d")private val random = new Random()private var pointer = -1def getUserID(): String = {pointer = pointer + 1if (pointer >= users.length) {pointer = 0users(pointer)} else {users(pointer)}}def click(): Double = {random.nextInt(10)}// bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --create --topic user_events --replication-factor 2 --partitions 2// bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --list// bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --describe user_events// bin/kafka-console-consumer.sh --zookeeper zk1:2181,zk2:2181,zk3:22181/kafka --topic test_json_basis_event --from-beginning
def main(args: Array[String]): Unit = {val topic = "user_events"val brokers = "localhost:9092"val props = new Properties()props.put("metadata.broker.list", brokers)props.put("serializer.class", "kafka.serializer.StringEncoder")val kafkaConfig = new ProducerConfig(props)val producer = new Producer[String, String](kafkaConfig)while (true) {// prepare event dataval event = new JSONObject()event.put("uid", getUserID).put("event_time", System.currentTimeMillis.toString).put("os_type", "Android").put("click_count", click)// produce event messageproducer.send(new KeyedMessage[String, String](topic, event.toString))println("Message sent: " + event)Thread.sleep(200)}}
}通过控制上面程序最后一行的时间间隔来控制模拟写入速度。
三、下面我们来讨论实现实时统计每个用户的点击次数，它是按照用户分组进行累加次数，逻辑比较简单，关键是在实现过程中要注意一些问题，如对象序列化等。先看实现代码，稍后我们再详细讨论，代码实现如下所示：
package com.sf.scalademo3import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka.KafkaUtils
import net.sf.json.JSONObject
import redis.clients.jedis.JedisPool
import org.apache.commons.pool2.impl.GenericObjectPoolConfigobject UserClickCountAnalytics {def main(args: Array[String]): Unit = {var masterUrl = "local[1]"if (args.length > 0) {masterUrl = args(0)}// Create a StreamingContext with the given master URLval conf = new SparkConf().setMaster(masterUrl).setAppName("UserClickCountStat")val ssc = new StreamingContext(conf, Seconds(5))// Kafka configurationsval topics = Set("user_events")val brokers = "localhost:9092"val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder")val dbIndex = 1val clickHashKey = "app::users::click"// Create a direct streamval kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)val events = kafkaStream.flatMap(line => {val data = JSONObject.fromObject(line._2)Some(data)})// Compute user click timesval userClicks = events.map(x => (x.getString("uid"), x.getInt("click_count"))).reduceByKey(_ + _)userClicks.foreachRDD(rdd => {rdd.foreachPartition(partitionOfRecords => {partitionOfRecords.foreach(pair => {val uid = pair._1val clickCount = pair._2val jedis = RedisClient.pool.getResourcejedis.select(dbIndex)jedis.hincrBy(clickHashKey, uid, clickCount)RedisClient.pool.returnResource(jedis)})})})ssc.start()ssc.awaitTermination()}
}上面代码使用了Jedis客户端来操作Redis，将分组计数结果数据累加写入Redis存储，如果其他系统需要实时获取该数据，直接从Redis实时读取即可。RedisClient实现代码如下所示：
package com.sf.scalademo3import redis.clients.jedis.JedisPool
import org.apache.commons.pool2.impl.GenericObjectPoolConfigobject RedisClient extends Serializable {val redisHost = "10.202.34.232"val redisPort = 6383val redisTimeout = 30000lazy val pool = new JedisPool(new GenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)lazy val hook = new Thread {override def run = {println("Execute hook thread: " + this)pool.destroy()}}sys.addShutdownHook(hook.run)
}上面代码我们分别在local[K]和Spark Standalone集群模式下运行通过。
如果我们是在开发环境进行调试的时候，也就是使用local[K]部署模式，在本地启动K个Worker线程来计算，这K个Worker在同一个JVM实例里，上面的代码默认情况是，如果没有传参数则是local[K]模式，所以如果使用这种方式在创建Redis连接池或连接的时候，可能非常容易调试通过，但是在使用Spark Standalone、YARN Client（YARN Cluster）或Mesos集群部署模式的时候，就会报错，主要是由于在处理Redis连接池或连接的时候出错了。我们可以看一下Spark架构，如图所示（来自官网）：
无论是在本地模式、Standalone模式，还是在Mesos或YARN模式下，整个Spark集群的结构都可以用上图抽象表示，只是各个组件的运行环境不同，导致组件可能是分布式的，或本地的，或单个JVM实例的。如在本地模式，则上图表现为在同一节点上的单个进程之内的多个组件；而在YARN Client模式下，Driver程序是在YARN集群之外的一个节点上提交Spark Application，其他的组件都运行在YARN集群管理的节点上。
在Spark集群环境部署Application后，在进行计算的时候会将作用于RDD数据集上的函数（Functions）发送到集群中Worker上的Executor上（在Spark Streaming中是作用于DStream的操作），那么这些函数操作所作用的对象（Elements）必须是可序列化的，通过Scala也可以使用lazy引用来解决，否则这些对象（Elements）在跨节点序列化传输后，无法正确地执行反序列化重构成实际可用的对象。上面代码我们使用lazy引用（Lazy Reference）来实现的，代码如下所示：
package com.sf.scalademo3import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.spark.streaming.kafka.KafkaUtils
import kafka.serializer.StringDecoder
import net.sf.json.JSONObject
import redis.clients.jedis.JedisPoolobject UserClickCountAnalytics2 {def main(args: Array[String]): Unit = {var masterUrl = "local[1]"if (args.length > 0) {masterUrl = args(0)}// Create a StreamingContext with the given master URLval conf = new SparkConf().setMaster(masterUrl).setAppName("UserClickCountStat")val ssc = new StreamingContext(conf, Seconds(5))// Kafka configurationsval topics = Set("user_events")val brokers = "localhost:9092"val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder")val dbIndex = 1val clickHashKey = "app::users::click"// Create a direct streamval kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)val events = kafkaStream.flatMap(line => {val data = JSONObject.fromObject(line._2)Some(data)})// Compute user click timesval userClicks = events.map(x => (x.getString("uid"), x.getInt("click_count"))).reduceByKey(_ + _)userClicks.foreachRDD(rdd => {rdd.foreachPartition(partitionOfRecords => {partitionOfRecords.foreach(pair => {/*** Internal Redis client for managing Redis connection {@link Jedis} based on {@link RedisPool}*/object InternalRedisClient extends Serializable {@transient private var pool: JedisPool = nulldef makePool(redisHost: String, redisPort: Int, redisTimeout: Int,maxTotal: Int, maxIdle: Int, minIdle: Int): Unit = {makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle, true, false, 10000)}def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,maxTotal: Int, maxIdle: Int, minIdle: Int, testOnBorrow: Boolean,testOnReturn: Boolean, maxWaitMillis: Long): Unit = {if (pool == null) {val poolConfig = new GenericObjectPoolConfig()poolConfig.setMaxTotal(maxTotal)poolConfig.setMaxIdle(maxIdle)poolConfig.setMinIdle(minIdle)poolConfig.setTestOnBorrow(testOnBorrow)poolConfig.setTestOnReturn(testOnReturn)poolConfig.setMaxWaitMillis(maxWaitMillis)pool = new JedisPool(poolConfig, redisHost, redisPort, redisTimeout)val hook = new Thread {override def run = pool.destroy()}sys.addShutdownHook(hook.run)}}def getPool: JedisPool = {assert(pool != null)pool}}// Redis configurationsval maxTotal = 10val maxIdle = 10val minIdle = 1val redisHost = "10.202.34.232"val redisPort = 6383val redisTimeout = 30000val dbIndex = 1InternalRedisClient.makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle)val uid = pair._1val clickCount = pair._2val jedis = InternalRedisClient.getPool.getResourcejedis.select(dbIndex)//原子操作--Redis HINCRBY命令用于增加存储在字段中存储由增量键哈希的数量。//如果键不存在,新的key被哈希创建。如果字段不存在,值被设置为0之前进行操作。
jedis.hincrBy(clickHashKey, uid, clickCount) InternalRedisClient.getPool.returnResource(jedis)})})})ssc.start()ssc.awaitTermination()}} 上面代码实现，得益于Scala语言的特性，可以在代码中任何位置进行class或object的定义，我们将用来管理Redis连接的代码放在了特定操作的内部，就避免了瞬态（Transient）对象跨节点序列化的问题。这样做还要求我们能够了解Spark内部是如何操作RDD数据集的，更多可以参考RDD或Spark相关文档。
在集群上，以Standalone模式运行，执行如下命令：
cd /usr/local/spark 2 ./bin/spark-submit --class org.shirdrn.spark.streaming.UserClickCountAnalytics --master spark://hadoop1:7077 --executor-memory 1G --total-executor-cores 2 ~/spark-0.0.SNAPSHOT.jar spark://hadoop1:7077
可以查看集群中各个Worker节点执行计算任务的状态，也可以非常方便地通过Web页面查看。
下面，看一下我们存储到Redis中的计算结果，如下所示：
有关更多关于Spark Streaming的详细内容，可以参考官方文档。
附录
这里，附上前面开发的应用所对应的依赖，以及打包Spark Streaming应用程序的Maven配置，以供参考。如果使用maven-shade-plugin插件，配置有问题的话，打包后在Spark集群上提交Application时候可能会报错Invalid signature file digest for Manifest main attributes。
demo2 Kafka+Spark Streaming+Redis实时计算整合实践 foreachRDD输出到redis相关推荐
- grafana计算不同时间的差值_大数据时代！如何基于Spark Streaming构建实时计算平台...
随着互联网技术的迅速发展,用户对于数据处理的时效性.准确性与稳定性要求越来越高,如何构建一个稳定易用并提供齐备的监控与预警功能的实时计算平台也成了很多公司一个很大的挑战. 自2015年携程实时计算平台 ...
- Flume+Kafka+Spark Streaming+MySQL实时日志分析
文章目录 项目背景 案例需求 一.分析 1.日志分析 二.日志采集 第一步.代码编辑 2.启动采集代码 三.编写Spark Streaming的代码 第一步 创建工程 第二步 选择创建Scala工程 ...
- 使用 Kafka 和 Spark Streaming 构建实时数据处理系统
使用 Kafka 和 Spark Streaming 构建实时数据处理系统 来源:https://www.ibm.com/developerworks,这篇文章转载自微信里文章,正好解决了我项目中的 ...
- 大数据Spark “蘑菇云”行动第76课： Kafka+Spark Streaming+Redis项目实战
大数据Spark "蘑菇云"行动第76课: Kafka+Spark Streaming+Redis项目实战 jedis插件 redis <dependency> ...
- Spark Streaming实现实时WordCount，DStream的使用，updateStateByKey(func)实现累计计算单词出现频率
一. 实战 1．用Spark Streaming实现实时WordCount 架构图: 说明:在hadoop1:9999下的nc上发送消息,消费端接收消息,然后并进行单词统计计算. * 2.安装并启动生 ...
- Spark Streaming 流式计算实战
这篇文章由一次平安夜的微信分享整理而来.在Stuq 做的分享,原文内容. 业务场景 这次分享会比较实战些.具体业务场景描述: 我们每分钟会有几百万条的日志进入系统,我们希望根据日志提取出时间以及用户名 ...
- Kafka+Spark Streaming如何保证exactly once语义
大数据技术与架构 点击右侧关注,大数据开发领域最强公众号! 暴走大数据 点击右侧关注,暴走大数据! 在Kafka.Storm.Flink.Spark Streaming等分布式流处理系统中(没错,Ka ...
- Kafka：ZK+Kafka+Spark Streaming集群环境搭建（九）安装kafka_2.11-1.1.0
如何搭建配置centos虚拟机请参考<Kafka:ZK+Kafka+Spark Streaming集群环境搭建(一)VMW安装四台CentOS,并实现本机与它们能交互,虚拟机内部实现可以上网.& ...
- Kafka：ZK+Kafka+Spark Streaming集群环境搭建（二十一）NIFI1.7.1安装
一.nifi基本配置 1. 修改各节点主机名,修改/etc/hosts文件内容. 192.168.0.120master192.168.0.121slave1192.168.0.122 slave2 ...
- Kafka：ZK+Kafka+Spark Streaming集群环境搭建（十二）VMW安装四台CentOS，并实现本机与它们能交互，虚拟机内部实现可以上网。...
Centos7出现异常:Failed to start LSB: Bring up/down networking. 按照<Kafka:ZK+Kafka+Spark Streaming集群环境搭 ...
最新文章
- 如何构建优质的推荐系统服务？| 技术头条
- R语言使用ggplot2包使用geom_boxplot函数绘制基础分组箱图（不同分组配置不同的箱体填充色+Brewer调色板）实战
- pytorch focalloss多分类 单分类
- 近90％公司亏损，泡沫破裂期将至：对于AI投资人的耐心还多吗
- 2018-北航-面向对象-前三次OO作业分析与小结
- 【机器学习PAI实践十二】机器学习算法基于信用卡消费记录做信用评分
- 【JavaScript】父子页面之间跨域通信的方法
- 学习office add-in
- git安装后找不见版本_无法安装最新版本的Gitlab
- 漏洞：WebRTC 泄漏用户IP
- QlikView线图高亮选择尺寸
- Answers To The Questions from GiGabyte
- react的setState——读《react小书》
- 【数据采集】-目前比较流行的几种数据采集方式
- C51与MDK共存 Keil5安装教程
- otn与stn网络_otn与stn网络_光通信网络
- teambition桌面客户端的妙用(绝密)
- 2020年度个税汇算清缴怎么办理？直接上干货！
- xml读取出现中文乱码
- vue使用天气API接口
热门文章
- 电脑已安装软件提取安装包_SPSS 24,软件安装包及安装教程
- vector容器v1、v2之间相互赋值的三种方法及易错点详解
- 估计理论(2)：Cramer-Rao下限
- 借助Haproxy_exporter实现对MarathonLb的流量和负载实例业务的可用状态监控-续
- 如何理解Beta分布和Dirichlet分布？
- 【树上贪心】Tree with Small Distances【codeforces-Round #506-div3-E】
- 矩阵变量的矩阵值函数微分公式 （原创）
- Ctftool：CTF漏洞利用工具
- Java多线程讲解，超详细！可获取相关笔记
- originos是鸿蒙系统吗,originos系统是安卓吗