文档章节

spark-streaming-kafka之createDirectStream模式

hblt-j
 hblt-j
发布于 2018/11/16 21:06
字数 1671
阅读 91
收藏 2

最近一直在用directstream方式消费kafka中的数据,特此总结,整个代码工程分为三个部分
一. 完整工程代码如下(某些地方特意做了说明, 这个代码的部分函数直接用的是spark-streaming-kafka-0.8_2.11)

package directStream

import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder
import kafka.common.TopicAndPartition

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kafka.common.TopicPartition
//import java.util._
import org.apache.spark.{SparkContext,SparkConf,TaskContext, SparkException}
import org.apache.spark.SparkContext._
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream._
import org.apache.spark.streaming.kafka.{KafkaUtils,HasOffsetRanges, OffsetRange,KafkaCluster}


import com.typesafe.config.ConfigFactory
import scalikejdbc._
import scala.collection.JavaConverters._

object SetupJdbc {
  def apply(driver: String, host: String, user: String, password: String): Unit = {
    Class.forName(driver)
    ConnectionPool.singleton(host, user, password)
  }
}
object SimpleApp{
  def main(args: Array[String]): Unit = {
  
    val conf = ConfigFactory.load // 加载工程resources目录下application.conf文件,该文件中配置了databases信息,以及topic及group消息
    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> conf.getString("kafka.brokers"),
      "group.id" -> conf.getString("kafka.group"),
      "auto.offset.reset" -> "smallest"
    )    
    val jdbcDriver = conf.getString("jdbc.driver")
    val jdbcUrl = conf.getString("jdbc.url")
    val jdbcUser = conf.getString("jdbc.user")
    val jdbcPassword = conf.getString("jdbc.password")

    val topic = conf.getString("kafka.topics")
    val group = conf.getString("kafka.group")

    val ssc = setupSsc(kafkaParams, jdbcDriver, jdbcUrl, jdbcUser, jdbcPassword,topic, group)()
    ssc.start()
    ssc.awaitTermination()
  }

  def createStream(taskOffsetInfo: Map[TopicAndPartition, Long], kafkaParams: Map[String, String], conf:SparkConf, ssc: StreamingContext, topics:String):InputDStream[_] = {
    // 若taskOffsetInfo 不为空, 说明这不是第一次启动该任务, database已经保存了该topic下该group的已消费的offset, 则对比kafka中该topic有效的offset的最小值和数据库保存的offset,去比较大作为新的offset.  
    if(taskOffsetInfo.size != 0){
        val kc = new KafkaCluster(kafkaParams)
        val earliestLeaderOffsets = kc.getEarliestLeaderOffsets(taskOffsetInfo.keySet) 
        if(earliestLeaderOffsets.isLeft)
          throw new SparkException("get kafka partition failed:") 
        val earliestOffSets = earliestLeaderOffsets.right.get

        val offsets = earliestOffSets.map(r => 
          new TopicAndPartition(r._1.topic, r._1.partition) -> r._2.offset.toLong)

        val newOffsets = taskOffsetInfo.map(r => {
            val t = offsets(r._1) 
            if (t > r._2) { 
              r._1 -> t
            } else {
              r._1 -> r._2
            }
          }
        ) 
        val messageHandler = (mmd: MessageAndMetadata[String, String]) => 1L 
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, Long](ssc, kafkaParams, newOffsets, messageHandler)    //val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
     } else {
        val topicSet = topics.split(",").toSet 
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams,topicSet)
   }
  }
    
  def setupSsc(
    kafkaParams: Map[String, String],
    jdbcDriver: String,
    jdbcUrl: String,
    jdbcUser: String,
    jdbcPassword: String,
    topics:String, 
    group:String
  )(): StreamingContext = {

    val conf = new SparkConf()
      .setMaster("mesos://10.142.113.239:5050")
      .setAppName("offset")
      .set("spark.worker.timeout", "500")
      .set("spark.cores.max", "10")
      .set("spark.streaming.kafka.maxRatePerPartition", "500")
      .set("spark.rpc.askTimeout", "600s")
      .set("spark.network.timeout", "600s")
      .set("spark.streaming.backpressure.enabled", "true")
      .set("spark.task.maxFailures", "1")
      .set("spark.speculationfalse", "false")



    val ssc = new StreamingContext(conf, Seconds(5))
    SetupJdbc(jdbcDriver, jdbcUrl, jdbcUser, jdbcPassword)  // connect to mysql 

    // begin from the the offsets committed to the database
    val fromOffsets = DB.readOnly { implicit session =>
      sql"select topic, part, offset from streaming_task where group_id=$group".
      map { resultSet =>
        new TopicAndPartition(resultSet.string(1), resultSet.int(2)) -> resultSet.long(3)
      }.list.apply().toMap
    }

    val stream = createStream(fromOffsets, kafkaParams, conf, ssc, topics)
    
    stream.foreachRDD { rdd =>
       if(rdd.count != 0){          
          // you task 
          val t = rdd.map(record => (record, 1))
          val results = t.reduceByKey {_+_}.collect
          

          // persist the offset into the database  
          val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
          DB.localTx { implicit session =>
               offsetRanges.foreach { osr => 
                 sql"""replace into streaming_task values(${osr.topic}, ${group}, ${osr.partition}, ${osr.untilOffset})""".update.apply()
                 if(osr.partition == 0){
                   println(osr.partition, osr.untilOffset)
                 }
               }
          }
      }
    }
    ssc
  }
}

二. 工程的resources文件下的有个application.conf配置文件,其配置如下

jdbc {
 driver = "com.mysql.jdbc.Driver"
 url = "jdbc:mysql://xxx.xxx.xxx.xxx:xxxx/xxxx"
 user = "xxxx"
 password = "xxxx"
}
kafka {
 topics = "xxxx"
 brokers = "xxxx.xxx.xxx.:xxx,xxx.xxx.xxx.xxx:9092,xxx.xxxx.xxx:xxxx"
 group = "xxxxxx"
}
jheckpointDir = "hdfs://xxx.xxx.xxx.xxx:9000/shouzhucheckpoint"
batchDurationMs = xxxx

三. 配置文件中可以看到, 我把offset 保存在 mysql里,这里我定义了一个table 名称为streaming_task, 表的结构信息如下:

+----------+--------------+------+-----+---------+-------+
| Field    | Type         | Null | Key | Default | Extra |
+----------+--------------+------+-----+---------+-------+
| topic    | varchar(100) | NO   | PRI | NULL    |       |
| group_id | varchar(50)  | NO   | PRI |         |       |
| part     | int(4)       | NO   | PRI | 0       |       |
| offset   | mediumtext   | YES  |     | NULL    |       |
+----------+--------------+------+-----+---------+-------+

部分解释如下:

一. 选用direct 的原因
官方为spark提供了两种方式来消费kafka中的数据, 高阶api由kafka自己来来维护offset, 有篇blog总结的比较好

第一种是利用 Kafka 消费者高级 API 在 Spark 的工作节点上创建消费者线程,订阅 Kafka 中的消息,数据会传输到 Spark 工作节点的执行器中,但是默认配置下这种方法在 Spark Job 出错时会导致数据丢失,如果要保证数据可靠性,需要在 Spark Streaming 中开启Write Ahead Logs(WAL),也就是上文提到的 Kafka 用来保证数据可靠性和一致性的数据保存方式。可以选择让 Spark 程序把 WAL 保存在分布式文件系统(比如 HDFS)中,

第二种方式不需要建立消费者线程,使用 createDirectStream 接口直接去读取 Kafka 的 WAL,将 Kafka 分区与 RDD 分区做一对一映射,相较于第一种方法,不需再维护一份 WAL 数据,提高了性能。读取数据的偏移量由 Spark Streaming 程序通过检查点机制自身处理,避免在程序出错的情况下重现第一种方法重复读取数据的情况,消除了 Spark Streaming 与 ZooKeeper/Kafka 数据不一致的风险。保证每条消息只会被 Spark Streaming 处理一次。以下代码片通过第二种方式读取 Kafka 中的数据:

在我在使用第一种方式的时候,如果数据量太大, 往往会出现报错,了解这这两种方式的不同后, 果断选用了第二种,

二. 引入KafkaCluster类的原因

引入KafkaCluster是为了在整个任务启动之前, 首先获取topic的有效的最旧offset. 这跟kafka的在实际的使用场景,大公司都是按时间删除kafka上数据有关,如果任务挂的时间太久,在还未能启动任务之前,database中保存的offset已经在kafak中失效,这时候为了最大程度的减少损失,只能从该topic的最旧数据开始消费..

三. 存入database的原因

看上面的代码,估计好多人也扒过KafkaCluster的源码, 这个类里面其实有一个setConsumerOffsets这样的方法�, 其实在处理过一个batch的数据后, 更新一下该topic下group的offset即可,但是还是在开始启动这个 job 的时候还得验证该offset否有效. 貌似这样还不用外部数据库,岂不方便? 其实这样做确实挺方便,
有些场景下这样做无可厚非, 但我觉得: 如果处理完数据,要写到外部数据库, 此时,如果能把写数据和写offset放在一个事务中(前提是这个数据库是支持事务), 那么就可以即可保证严格消费一次

四. conf 中两个特殊设置设置

为了确保task不会重复执行请设置下面两个参数:

  • spark.task.maxFailures=1, Task重试次数为1,即不重试
  • spark.speculation=false 关闭推测执行, 重点说下这个参数spark.speculation这个参数表示空闲的资源节点会不会尝试执行还在运行,并且运行时间过长的Task,避免单个节点运行速度过慢导致整个任务卡在一个节点上。这个参数最好设置为true。与之相配合可以一起设置的参数有spark.speculation.×开头的参数(设置spark.speculation=true将执行事件过长的节点去掉并重新分配任务而spark.speculation.interval用来设置执行间隔)


 

本文转载自:https://www.jianshu.com/p/2369a020e604

hblt-j
粉丝 24
博文 218
码字总数 73000
作品 0
海淀
架构师
私信 提问
Sparkstreaming and Kafka

简介 Kafka 0.10的Spark Streaming集成设计与0.8 Direct Stream方法类似。 它提供了简单的并行性,Kafka分区和Spark分区之间的1:1对应关系,以及对偏移量和元数据的访问。 但是,由于较新的...

舒运
2018/07/18
173
0
Spark Streaming整合kafka实战

kafka作为一个实时的分布式消息队列,实时的生产和消费消息,这里我们可以利用SparkStreaming实时计算框架实时地读取kafka中的数据然后进行计算。在spark1.3版本后,kafkaUtils里面提供了两个...

hblt-j
2018/11/15
339
0
Spark Streaming 中使用kafka低级api+zookeeper 保存 offset 并重用 以及 相关代码整合

在 Spark Streaming 中消费 Kafka 数据的时候,有两种方式分别是 1)基于 Receiver-based 的 createStream 方法和 2)Direct Approach (No Receivers) 方式的 createDirectStream 方法,详细...

cql252283126
2018/04/13
0
0
Spark实战 | Kafka与Spark Streaming的联姻

Kafka与Spark虽然没有很直接的必然关系,但是实际应用中却经常以couple的形式存在。如果你的Kafka的爱好者,流式计算框架Spark、Flink等也不得不去了解;如果你是Spark的爱好者,Kafka又或许...

HBase技术社区
02/28
0
0
kafka direct方式获取数据解析

---title: Spark Streaming基于kafka获取数据源码解析subtitle: kafka的direct方式获取数据description: kafkaRDD计算解析keywords: [spark,kafkaRDD,direct]date: 2019-01-08tags: [spark,k......

freeli
01/09
50
0

没有更多内容

加载失败,请刷新页面

加载更多

一次看懂 Https 证书认证

TLS > 传输层安全性协定 TLS(Transport Layer Security),及其前身安全套接层 SSL(Secure Sockets Layer)是一种安全协议,目的是为网际网路通信,提供安全及数据完整性保障。 如图,TLS...

极客收藏夹
37分钟前
4
0
https证书买哪家好?有哪些供应商

在选购https证书前除了要了解类型外,还需要了解https证书供应商,毕竟不同的供应商,提供的产品质量与服务也是有差异的。今天小编就为大家讲讲https证书供应商方面的内容,希望各位会喜欢。...

安信证书
39分钟前
5
0
Zuul 配置

概述:zuul底层是基于servlet,是由一系列的filter链构成。 1、路由配置 a、单例serverId映射 zuul: routes: client-a: path: /client/** serviceId: client-a 意思是...

java框架开发者
56分钟前
3
0
zk中FinalRequestProcessor解析

是处理器最后一个环节 FinalRequestProcessor implements RequestProcessor 处理器链最后一个环节处理事务和非事务请求最后一个环节 构造器 public FinalRequestProcessor(ZooKeeperServer z...

writeademo
57分钟前
4
0
Axios 详解

首先祝广大程序猿们节日快乐! 一、axios简介 基于promise,用于浏览器和node.js的http客户端 二、特点 支持浏览器和 node.js 支持 promise 能拦截请求和响应 能转换请求和响应数据 能取消请求...

张兴华ZHero
58分钟前
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部