Spark ´æ´¢Ðж¯Ëã×ÓÔ´Âë½âÎö

def saveAsHadoopFile(
    path: String,
    keyClass: Class[_],
    valueClass: Class[_],
    outputFormatClass: Class[_ <: OutputFormat[_, _]],
    conf: JobConf = new JobConf(self.context.hadoopConfiguration),
    codec: Option[Class[_ <: CompressionCodec]] = None): Unit = self.withScope {
  // Rename this as hadoopConf internally to avoid shadowing (see SPARK-2038).
  // 1. ÅäÖÃhadoopConf ½«key,value,outputÀàÐͽøÐÐÉèÖÃ
  val hadoopConf = conf
  hadoopConf.setOutputKeyClass(keyClass)
  hadoopConf.setOutputValueClass(valueClass)
  conf.setOutputFormat(outputFormatClass)
  // ÅäÖÃѹËõ
  for (c <- codec) {
    hadoopConf.setCompressMapOutput(true)
    hadoopConf.set("mapreduce.output.fileoutputformat.compress", "true")
    hadoopConf.setMapOutputCompressorClass(c)
    hadoopConf.set("mapreduce.output.fileoutputformat.compress.codec", c.getCanonicalName)
    hadoopConf.set("mapreduce.output.fileoutputformat.compress.type",
      CompressionType.BLOCK.toString)
  }
  // ÅäÖÃoutputµÄcommitter
  // Use configured output committer if already set
  if (conf.getOutputCommitter == null) {
    hadoopConf.setOutputCommitter(classOf[FileOutputCommitter])
  }

  // When speculation is on and output committer class name contains "Direct", we should warn
  // users that they may loss data if they are using a direct output committer.
  val speculationEnabled = self.conf.getBoolean("spark.speculation", false)
  val outputCommitterClass = hadoopConf.get("mapred.output.committer.class", "")
  if (speculationEnabled && outputCommitterClass.contains("Direct")) {
    val warningMessage =
      s"$outputCommitterClass may be an output committer that writes data directly to " +
        "the final location. Because speculation is enabled, this output committer may " +
        "cause data loss (see the case in SPARK-10063). If possible, please use an output " +
        "committer that does not have this behavior (e.g. FileOutputCommitter)."
    logWarning(warningMessage)
  }

  FileOutputFormat.setOutputPath(hadoopConf,
    SparkHadoopWriterUtils.createPathFromString(path, hadoopConf))
  // µ÷ÓÃsaveAsHadoopDataset
  saveAsHadoopDataset(hadoopConf)
}

def saveAsHadoopDataset(conf: JobConf): Unit = self.withScope {
  val config = new HadoopMapRedWriteConfigUtil[K, V](new SerializableJobConf(conf))
  SparkHadoopWriter.write(
    rdd = self,
    config = config)
}

def write[K, V: ClassTag](
    rdd: RDD[(K, V)],
    config: HadoopWriteConfigUtil[K, V]): Unit = {
  // Extract context and configuration from RDD.
  val sparkContext = rdd.context
  val commitJobId = rdd.id

// Set up a job.  ×¼±¸ºÍ´´½¨Ò»¸öcommiter Job
  val jobTrackerId =createJobTrackerID(new Date())
  val jobContext = config.createJobContext(jobTrackerId, commitJobId)
  config.initOutputFormat(jobContext)

  // Assert the output format/key/value class is set in JobConf.
  config.assertConf(jobContext, rdd.conf)

  val committer = config.createCommitter(commitJobId)
  committer.setupJob(jobContext)

  // Try to write all RDD partitions as a Hadoop OutputFormat.
  try {
    val ret = sparkContext.runJob(rdd, (context: TaskContext, iter: Iterator[(K, V)]) => {
      // SPARK-24552: Generate a unique "attempt ID" based on the stage and task attempt numbers.
      // Assumes that there won't be more than Short.MaxValue attempts, at least not concurrently.
      val attemptId = (context.stageAttemptNumber << 16) | context.attemptNumber
Ìá½»µ½Task¶ËÖ´ÐÐ
executeTask(
        context = context,
        config = config,
        jobTrackerId = jobTrackerId,
        commitJobId = commitJobId,
        sparkPartitionId = context.partitionId,
        sparkAttemptNumber = attemptId,
        committer = committer,
        iterator = iter)
    })
    // Ìá½»job
    committer.commitJob(jobContext, ret)
    logInfo(s"Job${jobContext.getJobID} committed.")
  } catch {
    case cause: Throwable =>
      logError(s"Aborting job${jobContext.getJobID}.", cause)
      committer.abortJob(jobContext)
      throw new SparkException("Job aborted.", cause)
  }
}

def saveAsHadoopFile[F <: OutputFormat[K, V]](
    path: String,
    codec: Class[_ <: CompressionCodec])(implicit fm: ClassTag[F]): Unit = self.withScope {
  val runtimeClass = fm.runtimeClass
  saveAsHadoopFile(path, keyClass, valueClass, runtimeClass.asInstanceOf[Class[F]], codec)
}

private[spark] def keyClass: Class[_] = kt.runtimeClass

private[spark] def valueClass: Class[_] = vt.runtimeClass

def saveAsTextFile(path: String): Unit = withScope {
  // same bytecodes for `saveAsTextFile`.
  val nullWritableClassTag =implicitly[ClassTag[NullWritable]]
  val textClassTag =implicitly[ClassTag[Text]]
  val r = this.mapPartitions { iter =>
    val text = new Text()
    iter.map { x =>
      text.set(x.toString)
      (NullWritable.get(), text)
    }
  }
  RDD.rddToPairRDDFunctions(r)(nullWritableClassTag, textClassTag, null)
    .saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
}

object HBaseWriteTest {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("HBaseWriteTest").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)

    val tableName = "XXX"
    val quorum = "localhost"
    val port = "2181"

    // ÅäÖÃÏà¹ØÐÅÏ¢
    val conf = HBaseUtils.getHBaseConfiguration(quorum,port,tableName)
    conf.set(TableOutputFormat.OUTPUT_TABLE,tableName)

    val jobConf = new JobConf()
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE,tableName)

    // дÈëÊý¾Ýµ½HBase
    val indataRDD = sc.makeRDD(Array("20180723_02,10","20180723_03,10","20180818_03,50"))

    val rdd = indataRDD.map(_.split(",")).map{arr => {
      val put = new Put(Bytes.toBytes(arr(0)))
      put.add(Bytes.toBytes("info"),Bytes.toBytes("clict_count"),Bytes.toBytes(arr(1)))
      (new ImmutableBytesWritable,put)
    }}.saveAsHadoopDataset(jobConf)

    sc.stop()
  }
}