This is because the DStream obj

昨天把项目代码重新整合了一下, 然后就报错了:

ERROR StreamingContext: Error starting the context, marking it as stopped
java.io.NotSerializableException: DStream checkpointing has been enabled but the DStreams with their functions are not serializable
com.gac.xs6.core.impl.EnAdapterImpl$


java.io.NotSerializableException: Object of org.apache.spark.streaming.kafka010.DirectKafkaInputDStream$DirectKafkaInputDStreamCheckpointData is being serialized  possibly as a part of closure of an RDD operation. This is because  the DStream object is being referred to from within the closure.  Please rewrite the RDD operation inside this DStream to avoid this.  This has been enforced to avoid bloating of Spark tasks  with unnecessary objects.

19/06/05 10:03:05 org.apache.spark.internal.Logging$class.logError(Logging.scala:91) ERROR Utils: Exception encountered
java.io.NotSerializableException: Object of org.apache.spark.streaming.dstream.DStreamCheckpointData is being serialized  possibly as a part of closure of an RDD operation. This is because  the DStream object is being referred to from within the closure.  Please rewrite the RDD operation inside this DStream to avoid this.  This has been enforced to avoid bloating of Spark tasks  with unnecessary objects.

原因可能是 mysql 数据库配置的序列化问题：

@Singleton
class MysqlConfiguration extends Serializable {
  private val config:             Config = ConfigFactory.load()
  lazy    val mysqlConf:          Config = config.getConfig("mysql")
  lazy    val mysqlJdbcUrl:       String = mysqlConf.getString("jdbcUrl")
  lazy    val mysqlJdbcDriver:    String = mysqlConf.getString("jdbcDriver")
  lazy    val dataSourceUser:     String = mysqlConf.getString("dataSource.user")
  lazy    val dataSourcePassword: String = mysqlConf.getString("dataSource.password")
  lazy    val dataSourceSize:     Int    = mysqlConf.getInt("dataSource.size")
}

但是我已经这么序列化了。

@Singleton
class NationApplication @Inject()(sparkConf:        SparkConfiguration,
                                  kafkaConf:        KafkaConfiguration,
                                  hbaseConf:        HbaseConfiguration,
                                  sparkContext:     NaSparkContext[SparkContext],
                                  mysqlConf:          MysqlConfiguration,
                                  source:           NaDStreamSource[Option[(String, String)]],
                                  adapter:          SourceDecode,
                                  sanitizer:        DataSanitizer,
                                  eventsExact:      EventsExact,
                                  eventsCheck:      EventsCheck,
                                  anomalyExact:     AnomalyExact,
                                  anomalyDetection: AnomalyDetection
                                  ) extends Serializable with Logging {

我在类里面添加了 MySQL 配置。然后在 eventStream.foreachRDD 里面引入

了 mysqlConf:

  def saveAnomaly(eventStream: DStream[(String, EventUpdate)]): Unit = {
    eventStream.foreachRDD((x: RDD[(String, EventUpdate)]) => {
      if (!x.isEmpty()) {
        x.foreachPartition { res: Iterator[(String, EventUpdate)] => {
           // val mysqlConf = new MysqlConfiguration
            val c      = new GacJdbcUtil(mysqlConf) // 获取 MySQL 数据库配置
            ...

结论：

在 foreachRDD 里面，使用 new 方法创建一个新的连接， 不使用类传递过来的配置