Spark Streaming：使用Google Pubsub进行自定义接收器kryo注册

Question

我正在使用Spark 2.0.2和Kryo​​序列化。

我正在尝试实现一个自定义接收器，用于将来自Google PubSub的消息提取到Spark Streaming中：

class PubSubReceiver(project: String, topic: String, subscription: String)
  extends Receiver[Array[Byte]](StorageLevel.MEMORY_AND_DISK_2) with Logging {
  val projectFullName = ProjectName.create(project)
  val topicName = TopicName.create(project, topic)
  val subscriptionName = SubscriptionName.create(project, subscription)
  val subscriber = Subscriber.defaultBuilder(subscriptionName, new receiver).build

  def onStart() {
    new Thread() {
      override def run() {
        subscriber.startAsync()
        //ensure subscriber is running as well as spark receiver
        while (subscriber.isRunning && !isStopped()) {
          logger.info(s"${subscriber.getSubscriptionName} receiver running")
          //sleep 10s
          Thread.sleep(10000)
        }
        logger.info(s"${subscriber.getSubscriptionName} receiver stopping")
      }
    }.start()
  }

  def onStop(): Unit = {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself if isStopped() returns false
  }

  private class receiver extends MessageReceiver {
    override def receiveMessage(message: PubsubMessage, consumer: AckReplyConsumer): Unit = {
      store(ArrayBuffer(message.getData.toByteArray), message.getAttributesMap)
    }
  }

}

但是，当运行使用此接收器的Spark作业时，似乎我必须序列化作业本身，这似乎不正确（然后将序列化Spark上下文）。

object PubSubStreamingIngestionJob extends App {
  //... setup

  lazy val ssc = new StreamingContext(spark.sparkContext, batchInterval)

  lazy val pubsubUnionStream =the stream
    ssc.receiverStream(new PubSubReceiver(projectName, topicName, subscriptionName))

  pubsubUnionStream.map( messageBytes => ...business logic... )

  ssc.start()
  ssc.awaitTermination()

}

抛出以下错误：

java.io.IOException: com.esotericsoftware.kryo.KryoException: java.lang.IllegalArgumentException: Class is not registered: com.c2fo.atlas.jobs.streaming.gcp.PubSubStreamingIngestionJob
Note: To register this class use: kryo.register(com.mycompany.package.PubSubStreamingIngestionJob.class);
Serialization trace:
classes (sun.misc.Launcher$AppClassLoader)
contextClassLoader (java.lang.Thread)
threads (java.lang.ThreadGroup)
parent (java.lang.ThreadGroup)
group (java.util.concurrent.Executors$DefaultThreadFactory)
val$backingThreadFactory (com.google.common.util.concurrent.ThreadFactoryBuilder$1)
threadFactory (java.util.concurrent.ScheduledThreadPoolExecutor)
e (java.util.concurrent.Executors$DelegatedScheduledExecutorService)
executor (com.google.cloud.pubsub.spi.v1.Subscriber)
subscriber (com.mycompany.package.PubSubReceiver)
array (scala.collection.mutable.WrappedArray$ofRef)

有没有更好的方法来实现这个？

Answer 1

问题是Subscriber实例需要是线程本地的，以防止整个闭包被序列化。

package org.apache.spark.streaming.gcp

import com.c2fo.atlas.util.LazyLogging
import com.google.cloud.pubsub.spi.v1._
import com.google.iam.v1.ProjectName
import com.google.pubsub.v1._
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

import scala.collection.mutable.ArrayBuffer

class PubSubReceiver(project: String, topic: String, subscription: String)
  extends Receiver[PubsubMessage](StorageLevel.MEMORY_AND_DISK_2) with LazyLogging{
  val projectFullName = ProjectName.create(project)
  val topicName = TopicName.create(project, topic)
  val subscriptionName = SubscriptionName.create(project, subscription)
  def onStart() {
    new Thread() {
      **//crucial change below**    
      val subscriber = Subscriber.defaultBuilder(subscriptionName, new receiver).build
      override def run() {
        subscriber.startAsync()
        //ensure subscriber is running as well as spark receiver
        while (subscriber.isRunning && !isStopped()) {
          logger.info(s"${subscriber.getSubscriptionName} receiver running")
          //sleep 10s
          Thread.sleep(10000)
        }
        logger.info(s"${subscriber.getSubscriptionName} receiver stopping")
      }
    }.start()
  }

  def onStop(): Unit = {
    // There is nothing much to do as the thread calling receive()
    // is designed to stop by itself if isStopped() returns false
  }

  class receiver extends MessageReceiver {
    override def receiveMessage(message: PubsubMessage, consumer: AckReplyConsumer): Unit = {
      store(ArrayBuffer(message), message.getAttributesMap)
    }
  }
}