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"),'	');
//需要写入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"),'	');
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对象，分隔符定为'	'制表符
        CSVReader reader =new CSVReader(new StringReader(s),'	');
        //以数组的形式返回每一行中的元素
        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)