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

import org.apache.spark.groupon.metrics.UserMetricsSystem
import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorRemoved, SparkListenerStageCompleted, SparkListenerTaskEnd}


class MetricsListener extends SparkListener {
  private lazy val executorRemovedMeter = UserMetricsSystem.meter("baryon.executorRemoved.rate")
  private lazy val failedStagesMeter = UserMetricsSystem.meter("baryon.failedStages.rate")
  private lazy val failedTasksMeter = UserMetricsSystem.meter("baryon.failedTasks.rate")

  override def onExecutorRemoved(executorRemoved: SparkListenerExecutorRemoved): Unit = {
    executorRemovedMeter.mark()
  }

  override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
    if (stageCompleted.stageInfo.failureReason.isDefined) {
      failedStagesMeter.mark()
    }
  }

  override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
    if (!taskEnd.taskInfo.successful) {
      failedTasksMeter.mark()
    }
  }
} 

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

import org.apache.spark.scheduler.SparkListener
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Assert._
import org.junit.runner.RunWith
import org.scalatest.{BeforeAndAfter, FunSuite}
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class ParallelizeTest extends FunSuite with BeforeAndAfter {

  var sparkContext: SparkContext = _

  before {
    sparkContext = SparkContextFactory.getContext()
  }

  after {
    if(sparkContext != null) {
      sparkContext.stop()
    }
  }

  test("Parallelize with 5 partitions") {

    val numberOfPartitions = 5
    val range = 1 to 10000
    val items = sparkContext.parallelize(range, numberOfPartitions)
    val filterItems = items.filter(item => item % 2 ==0)
    val result = filterItems.count()

    assertEquals(numberOfPartitions, items.partitions.length)
    assertEquals(range.end / 2, result)
  }

  test("Parallelize foreach partition") {

    sparkContext.addSparkListener(new SparkListener {

    })

    val numberOfPartitions = 5
    val range = 1 to 10000
    val items = sparkContext.parallelize(range, numberOfPartitions)

    assertEquals(5, items.partitions.length)
  }

} 

//设置package包名称以及导入依赖的类
package com.hazelcast.spark.connector.util

import com.hazelcast.spark.connector.util.ConnectionUtil.closeAll
import org.apache.spark.SparkContext
import org.apache.spark.scheduler.{SparkListener, SparkListenerJobEnd, SparkListenerJobStart}

object CleanupUtil {

  val jobIds: collection.mutable.Map[Int, Seq[Int]] = collection.mutable.Map[Int, Seq[Int]]()
  val cleanupJobRddName: String = "HazelcastResourceCleanupJob"

  def addCleanupListener(sc: SparkContext): Unit = {
    sc.addSparkListener(new SparkListener {
      override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
        this.synchronized {
          jobStart.stageInfos.foreach(info => {
            info.rddInfos.foreach(rdd => {
              if (!cleanupJobRddName.equals(rdd.name)) {
                val ids: Seq[Int] = info.rddInfos.map(_.id)
                val maybeIds: Option[Seq[Int]] = jobIds.get(jobStart.jobId)
                if (maybeIds.isDefined) {
                  jobIds.put(jobStart.jobId, ids ++ maybeIds.get)
                } else {
                  jobIds.put(jobStart.jobId, ids)
                }
              }
            })
          })
        }
      }

      override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
        this.synchronized {
          if (jobIds.contains(jobEnd.jobId)) {
            try {
              val workers = sc.getConf.getInt("spark.executor.instances", sc.getExecutorStorageStatus.length)
              val rddId: Option[Seq[Int]] = jobIds.get(jobEnd.jobId)
              if (rddId.isDefined) {
                sc.parallelize(1 to workers, workers).setName(cleanupJobRddName).foreachPartition(it ? closeAll(rddId.get))
              }
              jobIds -= jobEnd.jobId
            } catch {
              case e: Exception =>
            }
          }
        }
      }
    })
  }


} 

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

import com.typesafe.scalalogging.slf4j.{StrictLogging => Logging}
import org.apache.spark.SparkContext
import org.apache.spark.scheduler.{SparkListener, SparkListenerStageCompleted, SparkListenerStageSubmitted}
import org.apache.spark.sql.SparkSession
import org.karps.ops.{HdfsPath, HdfsResourceResult, SourceStamps}
import org.karps.structures.UntypedNodeJson

class KarpsListener(manager: Manager) extends SparkListener with Logging {
  override def onStageCompleted(stageCompleted: SparkListenerStageCompleted): Unit = {
    logger.debug(s"stage completed: $stageCompleted")
  }

  override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted): Unit = {
    logger.debug(s"stage submitted: $stageSubmitted")
  }
}



  def statusComputation(
      session: SessionId,
      computation: ComputationId): Option[BatchComputationResult] = {
    sessions.get(session).flatMap { ks =>
      ks.statusComputation(computation)
    }
  }

  def resourceStatus(session: SessionId, paths: Seq[HdfsPath]): Seq[HdfsResourceResult] = {
    sessions.get(session)
      .map(session => SourceStamps.getStamps(sparkSession, paths))
      .getOrElse(Seq.empty)
  }
}