自定义Spark Streaming接收器(Receivers)

Size: px

Start display at page:

Download "自定义Spark Streaming接收器(Receivers)"

鸿华杨
5 years ago
Views:

1 自定义 Spark Streaming 接收器 (Receivers) Spark Streaming 除了可以使用内置的接收器 (Receivers, 比如 Flume Kafka Kinesis file s 和 sockets 等 ) 来接收流数据, 还可以自定义接收器来从任意的流中接收数据开发者们可以自己实现 org.apache.spark.streaming.receiver.receiver 类来从其他的数据源中接收数据本文将介绍如何实现自定义接收器, 并且在 Spark Streaming 应用程序中使用我们可以用 Scala 或者 Java 来实现自定义接收器实现自定义接收器一切从实现 Receiver 开始所有的自定义接收器必须扩展 Receiver 抽象类, 并且实现其中两个方法 : 1 onstart(): 这个函数主要负责启动数据接收相关的事情 ; 2 onstop(): 这个函数主要负责停止数据接收相关的事情不管是 onstart() 和 onstop() 方法都不可以无限期地阻塞通常情况下,onStart() 会启动相关线程负责接收数据, 而 onstop() 会保证接收数据的线程被终止接收数据的线程也可以使用 Recei ver 类提供的 isstopped() 方法来检测是否可以停止接收数据数据一旦被接收, 这些数据可以通过调用 store(data) 方法存储在 Spark 中,store(data) 方法是由 Receiver 类提供的 Receiver 类提供了一系列的 store() 方法, 使得我们可以把一条一条地或者多条记录存储在 Spark 中值得注意的是, 使用某个 store() 方法实现接收器将会影响其可靠性和容错性, 这将在后面详细地讨论 1 / 11

$reporterror() 将会向 Dr iver 发送错误信息 ( 这些错误信息可以在 logs 和 UI 中看到 ), 而不停止或者重启接收器下面是一个从 socket 接收文本的自定义接收器它将 \n 作为一条记录的标志, 并且将这些数据存储在 Spark 中, 如果这些接收线程在连接或者接收数据的过程中出现错误, 接收器将重启并尝试再次连接, 详细的代码如下 :$

2 如果想及时了解 Spark Hadoop 或者 Hbase 相关的文章, 欢迎关注微信公共帐号 :iteblog_hadoop 接收线程中可能会出现任何异常, 这些异常都需要被捕获, 并且恰当地处理来避免接收器挂掉 restart() 将通过异步地调用 onstop() 和 onstart() 方法来重启接收器 stop() 将会调用 onstop() 方法并且中止接收器同样, reporterror() 将会向 Dr iver 发送错误信息 ( 这些错误信息可以在 logs 和 UI 中看到 ), 而不停止或者重启接收器下面是一个从 socket 接收文本的自定义接收器它将 \n 作为一条记录的标志, 并且将这些数据存储在 Spark 中, 如果这些接收线程在连接或者接收数据的过程中出现错误, 接收器将重启并尝试再次连接, 详细的代码如下 : ///////////////////////////////////////////////////////////////////// User: 过往记忆 Date: 2016 年 03 月 03 日 Time: 22:52:23 bolg: 本文地址 : 过往记忆博客, 专注于 hadoop hive spark shark flume 的技术博客, 大量的干货 2 / 11

3 过往记忆博客微信公共帐号 :iteblog_hadoop ///////////////////////////////////////////////////////////////////// class CustomReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) with Logging { def onstart() { // Start the thread that receives data over a connection new Thread("Socket Receiver") { override def run() { receive().start() def onstop() { // There is nothing much to do as the thread calling receive() // is designed to stop by itself if isstopped() returns false / Create a socket connection and receive data until receiver is stopped / private def receive() { var socket: Socket = null var userinput: String = null try { // Connect to host:port socket = new Socket(host, port) )) // Until stopped or connection broken continue reading val reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8" userinput = reader.readline() while(!isstopped && userinput!= null) { store(userinput) userinput = reader.readline() reader.close() socket.close() // Restart in an attempt to connect again when server is active again restart("trying to connect again") catch { case e: java.net.connectexception => // restart if could not connect to server restart("error connecting to " + host + ":" + port, e) case t: Throwable => // restart if there is any other error restart("error receiving data", t) 3 / 11

4 在 Spark Streaming 应用程序中使用自定义接收器自定义的接收器可以通过使用 streamingcontext.receiverstream() 方法来在 Spark Streaming 应用程序中使用这将使用自定义接收器接收到的数据来创建 input DStream, 如下 : // Assuming ssc is the StreamingContext val lines = ssc.receiverstream(new CustomReceiver(host, port)) val words = lines.flatmap(_.split(" "))... 完整的代码如下 : package org.apache.spark.examples.streaming import java.io.{bufferedreader, InputStream, InputStreamReader import java.net.socket import org.apache.spark.{logging, SparkConf import org.apache.spark.storage.storagelevel import org.apache.spark.streaming.{seconds, StreamingContext import org.apache.spark.streaming.receiver.receiver / Custom Receiver that receives data over a socket. Received bytes is interpreted as text and \n delimited lines are considered as records. They are then counted and printed. To run this on your local machine, you need to first run a Netcat server `$ nc -lk 9999` and then run the example `$ bin/runexample org.apache.spark.examples.streaming.customreceiver localhost 9999` / object CustomReceiver { def main(args: Array[String]) { if (args.length < 2) { System.err.println("Usage: CustomReceiver <hostname> <port>") 4 / 11

5 System.exit(1) StreamingExamples.setStreamingLogLevels() // Create the context with a 1 second batch size val sparkconf = new SparkConf().setAppName("CustomReceiver") val ssc = new StreamingContext(sparkConf, Seconds(1)) // Create a input stream with the custom receiver on target ip:port and count the // words in input stream of \n delimited text (eg. generated by 'nc') val lines = ssc.receiverstream(new CustomReceiver(args(0), args(1).toint)) val words = lines.flatmap(_.split(" ")) val wordcounts = words.map(x => (x, 1)).reduceByKey(_ + _) wordcounts.print() ssc.start() ssc.awaittermination() class CustomReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) with Logging { def onstart() { // Start the thread that receives data over a connection new Thread("Socket Receiver") { override def run() { receive().start() def onstop() { // There is nothing much to do as the thread calling receive() // is designed to stop by itself isstopped() returns false / Create a socket connection and receive data until receiver is stopped / private def receive() { var socket: Socket = null var userinput: String = null try { loginfo("connecting to " + host + ":" + port) socket = new Socket(host, port) loginfo("connected to " + host + ":" + port) val reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8" )) 5 / 11

6 userinput = reader.readline() while(!isstopped && userinput!= null) { store(userinput) userinput = reader.readline() reader.close() socket.close() loginfo("stopped receiving") restart("trying to connect again") catch { case e: java.net.connectexception => restart("error connecting to " + host + ":" + port, e) case t: Throwable => restart("error receiving data", t) 接收器的可靠性基于接收器的可靠性和容错性可以将接收器分为两大类 : 1 可靠的接收器 : 对于可靠的数据源, 它可以对发送出的数据进行确认, 可靠的接收器可以正确地通知数据源数据已经被接收并且已经可靠地存储到 Spark 中通常情况下, 实现这类接收器需要谨慎地考虑数据源确认语义 2 不可靠的接收器 : 不可靠的接收器并不发送确认信息到数据源这可以在哪些不支持确认的数据源中使用, 甚至在可靠的数据源中使用为了实现可靠的接收器, 你必须使用 store(multiple-records) 来存储数据调用这个 store 方法将会阻塞, 只有当所有的数据都被存储到 Spark 里面才会返回如果这个接收器配置的存储级别使用了副本机制 ( 默认就启用了 ), 这种情况下只有所有的副本都存储完成 store 才会返回这样才能保证数据被可靠地保存了, 然后接收器才可以通知数据源数据已经被接收这可以保证接收器在接收到一半数据的情况下也不会丢失数据, 因为接收器不会通知数据源, 因此数据源可以再次发送这些数据不可靠的接收器不需要实现这些逻辑它仅仅从数据源接收数据并且调用 store(single-recor d) 方法来一条一条地将数据存储在 Spark 中因为它并没有实现 store(multiplerecords) 方法的可靠性保证, 所以它有以下几点优势 : 1 系统来考虑如何恰当地将数据生成块 ; 2 如果速度限制被指定, 那么系统来考虑如何控制数据的接收速率 ; 3 因为上面 2 点优势, 实现不可靠的接收器比可靠的接收器要简单地多 6 / 11

7 下面对两类接收器的各自特点进行总结 : 1 不可靠的接收器 :(1) 比较容易实现 ;(2) 系统来提供块的生成和速率控制 ; 不需要提供容错保证, 但是可能在接收器挂掉的情况下丢失数据 2 可靠的接收器 :(1) 需要提供强容错保证, 能够保证零数据丢失 ;(2) 块生成和速率必须在接收器中实现 ;(3) 实现复杂度依赖于数据源确认机制的复杂度实现并使用基于 Actor 的自定义接收器用户自定义的 Akka Actor 同样可以被用于数据的接收 ActorHelper trait 可以使用在任何的 Akka actor 中, 并且可以使用 store(...) 方法把接收到的数据存储在 Spark 中 actor 的监管策略做相关的配置来处理异常等, 如下 : ///////////////////////////////////////////////////////////////////// User: 过往记忆 Date: 2016 年 03 月 03 日 Time: 22:52:23 bolg: 本文地址 : 过往记忆博客, 专注于 hadoop hive spark shark flume 的技术博客, 大量的干货过往记忆博客微信公共帐号 :iteblog_hadoop ///////////////////////////////////////////////////////////////////// class CustomActor extends Actor with ActorHelper { def receive = { case data: String => store(data) 然后使用下面的方法来使用这个自定义的 actor: // Assuming ssc is the StreamingContext val lines = ssc.actorstream[string](props(new CustomActor()), "CustomReceiver") 完整的代码如下 : package org.apache.spark.examples.streaming import scala.collection.mutable.linkedhashset 7 / 11

8 import scala.reflect.classtag import scala.util.random import akka.actor._ import com.typesafe.config.configfactory import org.apache.spark.sparkconf import org.apache.spark.streaming.{seconds, StreamingContext import org.apache.spark.streaming.akka.{actorreceiver, AkkaUtils case class SubscribeReceiver(receiverActor: ActorRef) case class UnsubscribeReceiver(receiverActor: ActorRef) / Sends the random content to every receiver subscribed with 1/2 second delay. / class FeederActor extends Actor { val rand = new Random() val receivers = new LinkedHashSet[ActorRef]() val strings: Array[String] = Array("words ", "may ", "count ") def makemessage(): String = { val x = rand.nextint(3) strings(x) + strings(2 - x) / A thread to generate random messages / new Thread() { override def run() { while (true) { Thread.sleep(500) receivers.foreach(_! makemessage).start() def receive: Receive = { case SubscribeReceiver(receiverActor: ActorRef) => println("received subscribe from %s".format(receiveractor.tostring)) receivers += receiveractor 8 / 11

9 case UnsubscribeReceiver(receiverActor: ActorRef) => println("received unsubscribe from %s".format(receiveractor.tostring)) receivers -= receiveractor / A sample actor as receiver, is also simplest. This receiver actor goes and subscribe to a typical publisher/feeder actor and receives [[org.apache.spark.examples.streaming.feederactor]] / class SampleActorReceiver[T](urlOfPublisher: String) extends ActorReceiver { lazy private val remotepublisher = context.actorselection(urlofpublisher) override def prestart(): Unit = remotepublisher! SubscribeReceiver(context.self) def receive: PartialFunction[Any, Unit] = { case msg => store(msg.asinstanceof[t]) override def poststop(): Unit = remotepublisher! UnsubscribeReceiver(context.self) / A sample feeder actor Usage: FeederActor <hostname> <port> <hostname> and <port> describe the AkkaSystem that Spark Sample feeder would start on. / object FeederActor { def main(args: Array[String]) { if (args.length < 2){ System.err.println("Usage: FeederActor <hostname> <port>\n") System.exit(1) val Seq(host, port) = args.toseq val akkaconf = ConfigFactory.parseString( s"""akka.actor.provider = "akka.remote.remoteactorrefprovider" akka.remote.enabled-transports = ["akka.remote.netty.tcp"] 9 / 11

10 akka.remote.netty.tcp.hostname = "$host" akka.remote.netty.tcp.port = $port """.stripmargin) val actorsystem = ActorSystem("test", akkaconf) val feeder = actorsystem.actorof(props[feederactor], "FeederActor") println("feeder started as:" + feeder) actorsystem.awaittermination() / A sample word count program demonstrating the use of plugging in Actor as Receiver Usage: ActorWordCount <hostname> <port> <hostname> and <port> describe the AkkaSystem that Spark Sample feeder is running on. To run this example locally, you may run Feeder Actor as `$ bin/run-example org.apache.spark.examples.streaming.feederactor localhost 9999` and then run the example `$ bin/runexample org.apache.spark.examples.streaming.actorwordcount localhost 9999` / object ActorWordCount { def main(args: Array[String]) { if (args.length < 2) { System.err.println( "Usage: ActorWordCount <hostname> <port>") System.exit(1) StreamingExamples.setStreamingLogLevels() val Seq(host, port) = args.toseq val sparkconf = new SparkConf().setAppName("ActorWordCount") // Create the context and set the batch size val ssc = new StreamingContext(sparkConf, Seconds(2)) / Following is the use of AkkaUtils.createStream to plug in custom actor as receiver An important point to note: Since Actor may exist outside the spark framework, It is thus user's responsibility to ensure the type safety, i.e type of data received and InputDStream 10 / 11

11 Powered by TCPDF ( 自定义 Spark Streaming 接收器 (Receivers) should be same. For example: Both AkkaUtils.createStream and SampleActorReceiver are parameterized to same type to ensure type safety. / val lines = AkkaUtils.createStream[String]( ssc, Props(classOf[SampleActorReceiver[String]], "akka.tcp://test@%s:%s/user/feederactor".format(host, port.toint)), "SampleReceiver") // compute wordcount lines.flatmap(_.split("\\s+")).map(x => (x, 1)).reduceByKey(_ + _).print() ssc.start() ssc.awaittermination() 本文翻译自 : Spark Streaming Custom Receivers : 本博客文章除特别声明, 全部都是原创! 转载本文请加上 : 转载自过往记忆 ( 本文链接 : () 11 / 11

使用Cassandra和Spark 2.0实现Rest API服务

使用Cassandra和Spark 2.0实现Rest API服务使用 Cassandra 和 Spark 2.0 实现 Rest API 服务在这篇文章中, 我将介绍如何在 Spark 中使用 Akkahttp 并结合 Cassandra 实现 REST 服务, 在这个系统中 Cassandra 用于数据的存储我们已经见识到 Spark 的威力, 如果和 Cassandra 正确地结合可以实现更强大的系统我们先创建一个 build.sbt 文件, 内容如下