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 >>

展开全文 >>

Spark架构原理

2018-06-09

前言

本文总结了Spark架构原理,其中主要包括五个组件：Driver、Master、Worker、Executor和Task，简要概括了每个组件是干啥的，并总结提交spark程序之后，这五个组件运行的详细步骤。

1、流程图

为了直观，就把流程图放在最前面了

more >>

展开全文 >>

Spark 持久化（cache和persist的区别）

2018-06-03

1、RDD 持久化

Spark 中一个很重要的能力是将数据持久化（或称为缓存），在多个操作间都可以访问这些持久化的数据。当持久化一个 RDD 时，每个节点的其它分区都可以使用 RDD 在内存中进行计算，在该数据上的其他 action 操作将直接使用内存中的数据。这样会让以后的 action 操作计算速度加快（通常运行速度会加速 10 倍）。缓存是迭代算法和快速的交互式使用的重要工具。

RDD 可以使用 persist() 方法或 cache() 方法进行持久化。数据将会在第一次 action 操作时进行计算，并缓存在节点的内存中。Spark 的缓存具有容错机制，如果一个缓存的 RDD 的某个分区丢失了，Spark 将按照原来的计算过程，自动重新计算并进行缓存。

在 shuffle 操作中（例如 reduceByKey），即便是用户没有调用 persist 方法，Spark 也会自动缓存部分中间数据。这么做的目的是，在 shuffle 的过程中某个节点运行失败时，不需要重新计算所有的输入数据。如果用户想多次使用某个 RDD，强烈推荐在该 RDD 上调用 persist 方法。

2、存储级别

每个持久化的 RDD 可以使用不同的存储级别进行缓存，例如，持久化到磁盘、已序列化的 Java 对象形式持久化到内存（可以节省空间）、跨节点间复制、以 off-heap 的方式存储在 Tachyon。这些存储级别通过传递一个 StorageLevel 对象给 persist() 方法进行设置。
详细的存储级别介绍如下：

MEMORY_ONLY : 将 RDD 以反序列化 Java 对象的形式存储在 JVM 中。如果内存空间不够，部分数据分区将不再缓存，在每次需要用到这些数据时重新进行计算。这是默认的级别。
MEMORY_AND_DISK : 将 RDD 以反序列化 Java 对象的形式存储在 JVM 中。如果内存空间不够，将未缓存的数据分区存储到磁盘，在需要使用这些分区时从磁盘读取。
MEMORY_ONLY_SER : 将 RDD 以序列化的 Java 对象的形式进行存储（每个分区为一个 byte 数组）。这种方式会比反序列化对象的方式节省很多空间，尤其是在使用 fast serializer时会节省更多的空间，但是在读取时会增加 CPU 的计算负担。
MEMORY_AND_DISK_SER : 类似于 MEMORY_ONLY_SER ，但是溢出的分区会存储到磁盘，而不是在用到它们时重新计算。
DISK_ONLY : 只在磁盘上缓存 RDD。
MEMORY_ONLY_2，MEMORY_AND_DISK_2，等等 : 与上面的级别功能相同，只不过每个分区在集群中两个节点上建立副本。
OFF_HEAP（实验中）: 类似于 MEMORY_ONLY_SER ，但是将数据存储在 off-heap memory，这需要启动 off-heap 内存。

注意，在 Python 中，缓存的对象总是使用 Pickle 进行序列化，所以在 Python 中不关心你选择的是哪一种序列化级别。python 中的存储级别包括 MEMORY_ONLY，MEMORY_ONLY_2，MEMORY_AND_DISK，MEMORY_AND_DISK_2，DISK_ONLY 和 DISK_ONLY_2 。
more >>

展开全文 >>