通用的load和save操作

• 对于Spark SQL的DataFrame来说，无论是从什么数据源创建出来的DataFrame，都有一些共同的load和save操作。load操作主要用于加载数据，创建出DataFrame；save操作，主要用于将DataFrame中的数据保存到文件中。

Java版本

DataFrame df = sqlContext.read().load("users.parquet");
df.select("name","favorite_color").write().save("namesAndFavColors.parquet");

java版本代码

package cn.spark.study.sql;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
public class GenericLoadSave {
public static void main(String[] args) {
SparkConf conf = new SparkConf()
.setAppName("GenericLoadSave");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
DataFrame usersDF = sqlContext.read().load("hdfs://spark1:9000/users.parquet");
usersDF.select("name","favorite_color").write()
.save("hdfs://spark1:9000/namesAndFavColors.parquet"); 
}
}

Scala版本

val df = sqlContext.read.load("users.parquet")
df.select("name","favorite_color").write.save("namesAndFavColors.parquet")

java版本代码

package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
object GenericLoadSave {
  def main(args: Array[String]){
    val conf = new SparkConf()
    .setAppName("GenericLoadSave")
    
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    
    val usersDF = sqlContext.read.load("hdfs://spark1:9000/users.parquet")
    usersDF.write.save("hdfs://spark1:9000/namesAndFavColors_scala")  
  }
}

手动指定数据源类型

• 也可以手动指定用来操作的数据源类型。数据源通常需要使用其全限定名来指定，比如parquet是org.apache.spark.sql.parquet。但是Spark SQL内置了一些数据源类型，比如json，parquet，jdbc等等。实际上，通过这个功能，就可以在不同类型的数据源之间进行转换了。比如将json文件中的数据保存到parquet文件中。默认情况下，如果不指定数据源类型，那么就是parquet。

Java版本

DataFrame df = sqlContext.read().format("json").load("people.json");
df.select("name","age").write().format("parquet").save("namesAndAges.parquet");

java版本代码

package cn.spark.study.sql;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
public class ManuallySpecifyOptions {
public static void main(String[] args) {
SparkConf conf = new SparkConf()
.setAppName("GenericLoadSave");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
DataFrame peopleDF = sqlContext.read().format("json")
.load("hdfs://spark1:9000/people.json");
peopleDF.select("name").write().format("parquet")
.save("hdfs://spark1:9000/peopleName_java");
}
}

Scala版本

val df = sqlContext.read.format("json").load("people.json")
df.select("name","age").write.format("parquet").save("namesAndAges.parquet")

Scala版本代码

package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
object ManuallySpecifyOptions {
  def main(args: Array[String]){
    val conf = new SparkConf()
    .setAppName("ManuallySpecifyOptions")
    
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc);
    
    val peopleDF = sqlContext.read.format("json").load("hdfs://spark1:9000/people.json")
    peopleDF.select("name").write.format("paruet").save("hdfs://spark1:9000/people.json")
  }
}

Save Mode

• Spark SQL对于save操作，提供了不同的save mode。主要用来处理，当目标位置，已经有数据时，应该如何处理。而且save操作并不会执行锁操作，并且不是原子的，因此是有一定风险出现脏数据的。

java版本代码

package cn.spark.study.sql;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SaveMode;
public class SaveModeTest {
@SuppressWarnings("deprecation")
public static void main(String[] args) {
SparkConf conf = new SparkConf()
.setAppName("GenericLoadSave");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
DataFrame peopleDF = sqlContext.read().format("json")
.load("hdfs://spark1:9000/people.json");
peopleDF.save("hdfs://spark1:9000/people_savemode_test", "json", SaveMode.Append);
}
}