SparkStreaming+Kafka 实现统计基于缓存的实时uv

2018-06-25

前言

本文利用SparkStreaming+Kafka实现实时的统计uv，即独立访客，一个用户一天内访问多次算一次，这个看起来要对用户去重，其实只要按照WordCount的思路，最后输出key的数量即可，所以可以利用SparkStreaming+Kafka 实现基于缓存的实时wordcount程序，这里稍加改动，如果uv数量增加的话就打印uv的数量(key的数量)。

1、数据

数据是我随机在kafka里生产的几条，用户以空格区分开（因为用的之前单词统计的程序）

2、kafka topic

首先在kafka建一个程序用到topic:KafkaUV

1	bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic KafkaUV

3、创建checkpoint的hdfs目录

我的目录为：/spark/dkl/kafka/UV_checkpoint

1	hadoop fs -mkdir -p /spark/dkl/kafka/UV_checkpoint

more >>

展开全文 >>

通过offsets.retention.minutes设置kafka offset的过期时间

2018-06-21

前言

本文记录博主如何设置kafka的offset过期时间并测试其效果

1、offsets.retention.minutes

通过修改offsets.retention.minutes的值即可改变kafka offset的过期时间，单位为分钟，改完之后需要重启kafka。具体的配置文件为$KAFKA_HOME/config/server.properties,原生的kafka配置文件里可能没有这个配置项，自己添加上即可，比如设置过期时间为一小时，那么按如下配置即可

1	offsets.retention.minutes=60

2、官方文档

网上有的博客说官网文档对于这个配置的说明有点错误，将offsets.retention.minutes错写成了offsets.topic.retention.minutes，但是我查看了一下，官方文档上并没有写错，可能是之前的版本写错了，而且很多博客按之前的版本写的，大家注意一下。官网文档地址http://kafka.apache.org/documentation/

3、ambari的bug

因本人用ambari管理大数据集群的各个组件，所以在界面上直接修改kafka的配置，在界面上查看kafka的配置offsets.retention.minutes为86400000，因为kafka offset默认过期时间为一天，那么根据这个86400000来看offsets.retention.minutes的单位为毫秒才对，所以一开始误认为单位为毫秒，所以修改配置后的时间设置的很大，导致一开始测试不成功，经过一点点的验证，发现单位实际上为分钟，而ambari上显示的86400000应该是个bug，因为kafka默认的配置文件里是没有这个配置项的，所以我估计ambari一开始也没有配置只是搜索的时候将其显示为86400000，而并没有真正的生效，只有将这个配置项修改之后，才会生效，并且单位为分钟（看了一下ambari的大部分默认时间单位都是毫秒~）。
后来在官网上看到offsets.retention.minutes的default为1440也证实了这一点。

4、测试效果

虽然本人的需求是将默认的一天的时间改长一点，但是时间长了测试太慢，所以将时间改短一点测试效果即可，测试代码见Spark Streamming+Kafka提交offset实现有且仅有一次,经过多次测试，得出结论，在修改重启之后，不管是新增加的topic还是之前的topic，只要是新保存的offset都会生效，而之前保存的offset，比如之前是一天才会删除，那么修改重启后，之前保存的offset还是会一天后才能删掉。
more >>

展开全文 >>

Spark Streaming+Kafka提交offset实现有且仅有一次(exactly-once)

2018-06-20

前言

本文讲Spark Streamming使用Direct方式读取Kafka，并在输出（存储）操作之后提交offset到Kafka里实现程序读写操作有且仅有一次，即程序重启之后之前消费并且输出过的数据不再重复消费，接着上次消费的位置继续消费Kafka里的数据。
Spark Streamming+Kafka官方文档：http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

1、提交offset的程序

根据官方文档可知，在spark代码里可以获取对应的offset信息，并且可以提交offset存储到kafka中。
代码：

package com.dkl.leanring.spark.kafka

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.TaskContext

object KafkaOffsetDemo {
  def main(args: Array[String]) {

    //创建sparkConf
    val sparkConf = new SparkConf().setAppName("KafkaOffsetDemo").setMaster("local[2]")
    // 创建StreamingContext batch size 为 1秒
    val ssc = new StreamingContext(sparkConf, Seconds(1))
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "ambari.master.com:6667", //kafka集群地址
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "KafkaOffsetDemo", //消费者组名
      "auto.offset.reset" -> "earliest", //当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      "enable.auto.commit" -> (false: java.lang.Boolean)) //如果是true，则这个消费者的偏移量会在后台自动提交
    val topics = Array("top1") //消费主题
    //创建DStream，返回接收到的输入数据
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams))
    // 打印获取到的数据，因为1秒刷新一次，所以数据长度大于0时才打印
    stream.foreachRDD(f => {

      if (f.count > 0) {
        println("=============================")
        println("打印获取到的kafka里的内容")
        f.foreach(f => {
          val value = f.value()
          println(value)

        })
        println("=============================")
        println("打印offset的信息")
        // offset
        val offsetRanges = f.asInstanceOf[HasOffsetRanges].offsetRanges

        //打印offset
        f.foreachPartition { iter =>
          val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
          println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
        }
        println("=============================")
        // 等输出操作完成后提交offset
        stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)

      }
    })
    //启动
    ssc.start()
    //等待停止
    ssc.awaitTermination()
  }
}

more >>

展开全文 >>

spark-submit提交Spark Streaming+Kafka程序

2018-06-19

前言

Spark Streaming本身是没有Kafka相关的jar包和API的，如果想利用Spark Streaming获取Kafka里的数据，需要自己将依赖添加SBT或Maven项目中，添加依赖更新项目之后，就可以在Eclipse等IDE里直接运行Spark Streamming+Kafka的程序了，可参考Spark Streaming连接Kafka入门教程，但是如果需要在集群通过spark-submit提交jar包的方式来运行程序的话，会抛出异常：

1	Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/StringDeserializer

这是因为Spark本身没有Kafka相关的jar，所以需要将与Kafka相关的jar添加Spark环境中

1、相关jar包

总共需要两个jar，可以在SBT和Maven下载的目录里找到，分别为org.apache.spark/spark-streaming-kafka-0-10_2.11/jars/spark-streaming-kafka-0-10_2.11-2.3.0.jar和org.apache.kafka/kafka-clients/jars/kafka-clients-0.10.0.1.jar,具体的名字会因你的kafka版本和spark版本而有所不同
more >>

展开全文 >>

SparkStreaming+Kafka 实现基于缓存的实时wordcount程序

2018-06-14

前言

本文利用SparkStreaming和Kafka实现基于缓存的实时wordcount程序，什么意思呢，因为一般的SparkStreaming的wordcount程序比如官网上的，只能统计最新时间间隔内的每个单词的数量，而不能将历史的累加起来，本文是看了教程之后，自己实现了一下kafka的程序，记录在这里。其实没什么难度，只是用了一个updateStateByKey算子就能实现，因为第一次用这个算子，所以正好学习一下。

1、数据

数据是我随机在kafka里生产的几条，单词以空格区分开

2、kafka topic

首先在kafka建一个程序用到topic:UpdateStateBykeyWordCount

1	bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic UpdateStateBykeyWordCount

3、创建checkpoint的hdfs目录

我的目录为：/spark/dkl/kafka/wordcount_checkpoint

1	hadoop fs -mkdir -p /spark/dkl/kafka/wordcount_checkpoint

more >>

展开全文 >>