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 jackage of someone is 21（2）fastJsonfastJson中也要借助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)