1.  load&save

def parallelize[T: ClassTag](
     seq: Seq[T],
     numSlices: Int = defaultParallelism): RDD[T] = withScope {
   assertNotStopped()
   new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
 }

val rdd: RDD[Int] = sc.makeRDD(Array(1,2,3,4))
 rdd.collect

def makeRDD[T: ClassTag](
     seq: Seq[T],
     numSlices: Int = defaultParallelism): RDD[T] = withScope {
   parallelize(seq, numSlices)
 }

def textFile(
     path: String,
     minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
   hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
     minPartitions).map(pair => pair._2.toString).setName(path)
 }

def saveAsTextFile(path: String): Unit = withScope {
     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)
 }

def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit = withScope {
   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, codec)
 }

def saveAsObjectFile(path: String): Unit = withScope {
   this.mapPartitions(iter => iter.grouped(10).map(_.toArray))
     .map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x))))
     .saveAsSequenceFile(path)
 }

def objectFile[T: ClassTag](
     path: String,
     minPartitions: Int = defaultMinPartitions): RDD[T] = withScope {
   assertNotStopped()
   sequenceFile(path, classOf[NullWritable], classOf[BytesWritable], minPartitions)
     .flatMap(x => Utils.deserialize[Array[T]](x._2.getBytes,Utils.getContextOrSparkClassLoader))
 }

源码：

def saveAsSequenceFile(
     path: String,
     codec: Option[Class[_ <: CompressionCodec]] = None): Unit = self.withScope {
   def anyToWritable[U <% Writable](u: U): Writable = u
   
   val convertKey = self.keyClass != keyWritableClass
   val convertValue = self.valueClass != valueWritableClass
   
   logInfo("Saving as sequence file of type (" + keyWritableClass.getSimpleName + "," +
     valueWritableClass.getSimpleName + ")" )
   val format = classOf[SequenceFileOutputFormat[Writable, Writable]]
   val jobConf = new JobConf(self.context.hadoopConfiguration)
   if (!convertKey && !convertValue) {
     self.saveAsHadoopFile(path, keyWritableClass, valueWritableClass, format, jobConf, codec)
   } else if (!convertKey && convertValue) {
     self.map(x => (x._1, anyToWritable(x._2))).saveAsHadoopFile(
       path, keyWritableClass, valueWritableClass, format, jobConf, codec)
   } else if (convertKey && !convertValue) {
     self.map(x => (anyToWritable(x._1), x._2)).saveAsHadoopFile(
       path, keyWritableClass, valueWritableClass, format, jobConf, codec)
   } else if (convertKey && convertValue) {
     self.map(x => (anyToWritable(x._1), anyToWritable(x._2))).saveAsHadoopFile(
       path, keyWritableClass, valueWritableClass, format, jobConf, codec)
   }
 }

def sequenceFile[K, V]
      (path: String, minPartitions: Int = defaultMinPartitions)
      (implicit km: ClassTag[K], vm: ClassTag[V],
       kcf: () => WritableConverter[K], vcf: () => WritableConverter[V]): RDD[(K, V)] = {
   withScope {
     assertNotStopped()
     val kc = clean(kcf)()
     val vc = clean(vcf)()
     val format = classOf[SequenceFileInputFormat[Writable, Writable]]
     val writables = hadoopFile(path, format,
       kc.writableClass(km).asInstanceOf[Class[Writable]],
       vc.writableClass(vm).asInstanceOf[Class[Writable]], minPartitions)
     writables.map { case (k, v) => (kc.convert(k), vc.convert(v)) }
   }
 }