//设置package包名称以及导入依赖的类
package org.hammerlab.hadoop.kryo

import java.io.{ DataInputStream, DataOutputStream }

import com.esotericsoftware.kryo.io.{ Input, Output }
import com.esotericsoftware.kryo.{ Kryo, Serializer }
import org.apache.hadoop.io.Writable

class WritableSerializer[T <: Writable](ctorArgs: Any*) extends Serializer[T] {
  override def read(kryo: Kryo, input: Input, clz: Class[T]): T = {
    val t = clz.newInstance()
    t.readFields(new DataInputStream(input))
    t
  }

  override def write(kryo: Kryo, output: Output, t: T): Unit = {
    t.write(new DataOutputStream(output))
  }
} 

//设置package包名称以及导入依赖的类
package com.groupon.dse.mezzanine

import com.groupon.dse.kafka.common.WrappedMessage
import com.groupon.dse.mezzanine.converter.WritableConverter
import com.groupon.dse.mezzanine.extractor.{TopicAndEvent, WrappedMessageExtractor}
import com.groupon.dse.mezzanine.outputformat.StagingFileOutputFormat
import com.groupon.dse.mezzanine.partitioner.Partitioner
import com.groupon.dse.mezzanine.util.HDFSWriter
import com.groupon.dse.spark.plugins.ReceiverPlugin
import org.apache.hadoop.io.Writable
import org.apache.spark.rdd.RDD

import scala.concurrent.duration.Duration
import scala.reflect.ClassTag


  override def execute(messages: RDD[WrappedMessage]): Unit = {
    val topics = messages.mapPartitions(partitionIterator => {
      if (partitionIterator.nonEmpty) {
        Iterator((partitionIterator.next().topic, null))
      } else {
        Iterator.empty
      }
    }).reduceByKeyLocally((x, y) => x).keys
    val topicToStagingPath = topics.map(topic => topic -> partitioner.relativeStagingDirectory(topic)).toMap

    val topicEventRDD = extractor.extract(messages)

    val pathEventRDD = topicEventRDD.mapPartitions(partitionIterator => {
      if (partitionIterator.nonEmpty) {
        val firstTopicAndEvent = partitionIterator.next()
        val stagingPath = topicToStagingPath.get(firstTopicAndEvent.topic).get

        (Iterator(firstTopicAndEvent) ++ partitionIterator).map(topicAndEvent => {
          (stagingPath, converter.convert(topicAndEvent))
        })
      } else {
        Iterator.empty
      }
    }).cache()

    HDFSWriter.multiWriteToHDFSwithRetry(
      pathEventRDD,
      partitioner.stagingRoot,
      classOf[StagingFileOutputFormat[K, V]],
      retryInterval
    )
  }
} 

//设置package包名称以及导入依赖的类
package com.groupon.dse.mezzanine.compactor

import com.groupon.dse.mezzanine.partitioner.Partitioner
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.io.Writable
import org.apache.spark.SparkContext
import org.apache.spark.groupon.metrics.UserMetricsSystem
import org.apache.spark.streaming.scheduler.{StreamingListener, StreamingListenerBatchCompleted}


class CompactorListener[K <: Writable, V <: Writable](val sparkContext: SparkContext,
                                                      val fs: FileSystem,
                                                      val partitioner: Partitioner,
                                                      val compactor: Compactor[K, V]) extends StreamingListener {

  override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    // Get the leaf directories where staging files are written to for the topics we consume
    val stagingLeafDirs = partitioner.stagingLeafDirectories(fs).filter(path => {
      path.getName.startsWith(Partitioner.KeyPrefix)
    })

    stagingLeafDirs.par.foreach(stagingPath => {
      val key = partitioner.keyForStagingDirectory(stagingPath)
      val filesToCompact = compactor.getFilesToCompact(stagingPath, batchCompleted.batchInfo.processingEndTime.get)
      if (filesToCompact.nonEmpty) {
        val outputPath = partitioner.outputDirectory(key)
        UserMetricsSystem.timer(s"mezzanine.write.time.output.$key").time({
          compactor.saveToOutputDir(filesToCompact, outputPath)
        })
        filesToCompact.foreach(fs.delete(_, false))
      }
    })
  }
}