spark streaming - kafka updateStateByKey 统计用户消费金额

场景

餐厅老板想要统计每个用户来他的店里总共消费了多少金额，我们可以使用updateStateByKey来实现

从kafka接收用户消费json数据，统计每分钟用户的消费情况，并且统计所有时间所有用户的消费情况(使用updateStateByKey来实现)

数据格式

{"user":"zhangsan","payment":8}
{"user":"wangwu","payment":7}
....

往kafka写入消息(kafka producer)

package producerimport java.util.Propertiesimport kafka.javaapi.producer.Producer
import kafka.producer.{KeyedMessage, ProducerConfig}
import org.codehaus.jettison.json.JSONObject
import scala.util.Randomobject KafkaProducer extends App{ //所有用户
private val users = Array(
"zhangsan", "lisi",
"wangwu", "zhaoliu") private val random = new Random() //消费的金额（0-9）
def payMount() : Double = {
random.nextInt(10)
} //随机获得用户名称
def getUserName() : String = {
users(random.nextInt(users.length))
} //kafka参数
val topic = "user_payment"
val brokers = "192.168.6.55:9092,192.168.6.56: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) {
// 创建json串
val event = new JSONObject()
event
.put("user", getUserName())
.put("payment", payMount) // 往kafka发送数据
producer.send(new KeyedMessage[String, String](topic, event.toString))
println("Message sent: " + event) //每隔200ms发送一条数据
Thread.sleep(200)
}
}


使用spark Streaming处理数据

import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{StreamingContext, Seconds}
import org.apache.spark.{SparkContext, SparkConf}
import net.liftweb.json._object UpdateStateByKeyTest { def main (args: Array[String]) { def functionToCreateContext(): StreamingContext = {
//创建streamingContext
val conf = new SparkConf().setAppName("test").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(60)) //将数据进行保存（这里作为演示，生产中保存在hdfs）
ssc.checkpoint("checkPoint") val zkQuorum = "192.168.6.55:2181,192.168.6.56:2181,192.168.6.57:2181"
val consumerGroupName = "user_payment"
val kafkaTopic = "user_payment"
val kafkaThreadNum = 1 val topicMap = kafkaTopic.split(",").map((_, kafkaThreadNum.toInt)).toMap //从kafka读入数据并且将json串进行解析
val user_payment = KafkaUtils.createStream(ssc, zkQuorum, consumerGroupName, topicMap).map(x=>{
parse(x._2)
}) //对一分钟的数据进行计算
val paymentSum = user_payment.map(jsonLine =>{
implicit val formats = DefaultFormats
val user = (jsonLine \ "user").extract[String]
val payment = (jsonLine \ "payment").extract[String]
(user,payment.toDouble)
}).reduceByKey(_+_) //输出每分钟的计算结果
paymentSum.print() //将以前的数据和最新一分钟的数据进行求和
val addFunction = (currValues : Seq[Double],preVauleState : Option[Double]) => {
val currentSum = currValues.sum
val previousSum = preVauleState.getOrElse(0.0)
Some(currentSum + previousSum)
} val totalPayment = paymentSum.updateStateByKey[Double](addFunction) //输出总计的结果
totalPayment.print() ssc
} //如果"checkPoint"中存在以前的记录，则重启streamingContext,读取以前保存的数据，否则创建新的StreamingContext
val context = StreamingContext.getOrCreate("checkPoint", functionToCreateContext _) context.start()
context.awaitTermination()
}
}


运行结果节选

//-----------第n分钟的结果------------------//1分钟结果
-------------------
(zhangsan,23.0)
(lisi,37.0)
(wangwu,31.0)
(zhaoliu,34.0)
-------------------//总和结果
(zhangsan,101.0)
(lisi,83.0)
(wangwu,80.0)
(zhaoliu,130.0)//-----------第n+1分钟的结果------------------//1分钟结果
-------------------
(zhangsan,43.0)
(lisi,16.0)
(wangwu,21.0)
(zhaoliu,54.0)
-------------------
//总和结果
-------------------
(zhangsan,144.0)
(lisi,99.0)
(wangwu,101.0)
(zhaoliu,184.0)
-------------------

后记

下一片文章为统计不同时间段用户平均消费金额，消费次数，消费总额等指标。

点击这里