spark中數據的讀取與保存

1.文本文件

(1)讀取文本文件

JavaRDD<String> input =sc.textFile(dir)

(2)保存文本文件

result.saveAsTextFile(dir);

2.Json

 （1）gson

①Gson中需要創建JavaBean實體類來說明json的格式。

以下是javaBean的示例程序，Person，其中有兩個字段，分別代表姓名和年齡。

public class Person {
    private String name;
    private int age;

    public Person(String name, int age) {
        this.name = name;
        this.age = age;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

② 通過Person實例創建Json String

final Gson gson=new Gson();
Person jack =new Person("jack",21);
String json=gson.toJson(jack);
System.out.println(json);

輸出：

{"name":"jack","age":21}

③  通過Json String 重建javaBean實例

try {
    Person someone=gson.fromJson(json,Person.class);

    System.out.println("name if someone is"+someone.getName());
    System.out.println("age of someone is"+someone.getAge());

} catch (JsonSyntaxException e) {
    e.printStackTrace();
}

輸出：

name if someone is jack

age of someone is 21

（2）fastJson

fastJson中也要借助javaBean完成json String的創建和解析，我們依然采用上文Person類作為javaBean的定義。

①  通過Person實例創建json String

Person jack =new Person("jack",21);
String json = JSON.toJSONString(jack);
System.out.println(json);

②  解析json String

Person someone=JSON.parseObject(json,Person.class);

* 注意，fastJson發序列化json String javaBean定義中必須加上默認構造函數，就像Person類的定義中需要對默認構造函數的定義，否則會報錯：

public Person(){

}

3.逗號分隔值與制表符分隔值

(1)創建csv文件

//創建CSVWriter,文件路徑為c://test.csv,分隔符為制表符
CSVWriter writer =new CSVWriter(new FileWriter("C://test.csv"),'\t');
//需要寫入csv文件的一行的三個String
String [] line={"aaa","bbb","ccc"};
//寫入這一行
writer.writeNext(line);
writer.close();

輸出：

"aaa" "bbb" "ccc"

如果要連續輸出幾行的數據，可以這么做：

//創建CSVWriter,文件路徑為c://test.csv,分隔符為制表符
CSVWriter writer =new CSVWriter(new FileWriter("C://test.csv"),'\t');
List<String[]> lines=new ArrayList<String[]>();
lines.add(new String []{"hhh","ggg","hhh"});
lines.add(new String[]{"xxx","yyy","zzz"});
writer.writeAll(lines);
writer.close();

輸出：

"hhh" "ggg" "hhh"

"xxx" "yyy" "zzz"

(2)讀取csv文件

現在我們用OpenCsv來讀取我們剛才創建的csv文件，實現如下：

//讀取csv文件，以行為單位，保存在lines中
JavaRDD<String> lines = sc.textFile("C://test.csv");
//定義如何將一行中的元素讀取出來，以String[]的形式返回
class ParseLine  implements  Function<String,String[]>{

    public String[] call(String s) throws Exception {
        //新建CSVReader對象，分隔符定為'\t'制表符
        CSVReader reader =new CSVReader(new StringReader(s),'\t');
        //以數組的形式返回每一行中的元素
        return reader.readNext();
    }
}
//利用ParseLine，轉化處理lines
JavaRDD<String[]> results=lines.map(
       new ParseLine()
);
//遍歷輸出results中的內容
for(String  s []: results.collect() ){
    System.out.println("this is the elements of one line!");
    for(String str:s)
        System.out.println(str);
}

4.SequenceFile

(1)簡介

    SequenceFile是由沒有相對關系結構的鍵值對組成的常用hadoop格式。SequenceFile文件有同步標記，Spark可以用它來定位到文件的某個點，然后再與記錄的邊界對齊。這可以讓Spark 使用多個節點高效地並行讀取SequenceFile文件。SequenceFile也是Hadoop MapReduce作業中常用的輸入輸出格式。

(2)以SequenceFile的格式保存PairRDD

//新建PairRDD<String,Integer>
JavaPairRDD<String,Integer> data =sc
        .parallelizePairs(Arrays.asList(new Tuple2<String, Integer>("zhouyang", 1),
                new Tuple2<String, Integer>("jack", 2),
                new Tuple2<String, Integer>("bob", 3)));
//將PairRDD<String,Integer> 轉換為hadoop io中對應的格式 PairRDD<Text,IntWritable> 
JavaPairRDD<Text,IntWritable>  result =data.mapToPair(
        new PairFunction<Tuple2<String, Integer>, Text, IntWritable>() {
            public Tuple2<Text, IntWritable> call(Tuple2<String, Integer> record) throws Exception {
                return new Tuple2(new Text(record._1()),new IntWritable(record._2()));
            }
        }
);
//將result以SequenceFile保存在指定目錄下
result.saveAsHadoopFile("C://hadoop.file",Text.class,IntWritable.class, SequenceFileOutputFormat.class);

(3)讀取SequenceFile中的PairRDD，在(2)中創建的hadoop.file文件

JavaPairRDD<Text,IntWritable> input=sc.sequenceFile("c://hadoop.file",Text.class,IntWritable.class);
JavaPairRDD<String,Integer> results =input.mapToPair(
        new PairFunction<Tuple2<Text, IntWritable>, String, Integer>() {
            public Tuple2<String, Integer> call(Tuple2<Text, IntWritable> record) throws Exception {
                return new Tuple2<String,Integer>(record._1().toString(),record._2().get());
            }
        }
);
for(Tuple2<String,Integer> tuple: results.collect())
    System.out.println(tuple._1()+"->" +tuple._2());

(4)對象文件

    對象文件允許存儲只包含值的RDD。和普通的SequenceFile不同，對於同樣的對象，對象文件的輸出和Hadoop輸出不一樣。

① 將PairRDD保存為對象文件

JavaPairRDD<String,Integer> data =sc
        .parallelizePairs(Arrays.asList(new Tuple2<String, Integer>("zhouyang", 1),
 new Tuple2<String, Integer>("jack", 2),
 new Tuple2<String, Integer>("bob", 3)));
data.saveAsObjectFile("c://object.file");

②  讀取對象文件的PairRDD,因為對象文件只存儲包含值的RDD,所以讀出時，應為JavaRDD<Tuple2<String,Integer>>格式

JavaRDD<Tuple2<String,Integer>> input=sc.objectFile("c://object.file");
for(Tuple2<String,Integer> tuple:input.collect()){
    System.out.println(tuple._1()+" -> "+tuple._2());
}

5.  protocol buffer                                                                                                                                                                                     

 

來自為知筆記(Wiz)