1.常用的方式通过sparksession读取外部文件或者数据生成dataset(这里就不讲了）
注： 生成Row对象的方法提一下：RowFactory.create(x,y,z)，取Row中的数据使用row.getAs("列名")来获取对应的列值或者row.getInt(0),row.getString(1)（但这个要注意顺序）
2.通过调用createDataFrame生成Dataset
通过反射的方式将非json格式的RDD转换成DataFrame（不建议使用）
自定义类要可序列化
自定义类的访问级别是Public
RDD转成DataFrame后会根据映射将字段按Assci码排序
将DataFrame转换成RDD时获取字段两种方式,一种是df.getInt(0)下标获取（不推荐使用），另一种是df.getAs(“列名”)获取（推荐使用）
关于序列化问题:
1.反序列化时serializable 版本号不一致时会导致不能反序列化。
2.子类中实现了serializable接口，父类中没有实现，父类中的变量不能被序列化,序列化后父类中的变量会得到null。
注意：父类实现serializable接口,子类没有实现serializable接口时，子类可以正常序列化
3.被关键字transient修饰的变量不能被序列化。
4.静态变量不能被序列化，属于类，不属于方法和对象，所以不能被序列化。
另外：一个文件多次writeObject时，如果有相同的对象已经写入文件，那么下次再写入时，只保存第二次写入的引用，读取时，都是第一次保存的对象。
/**方法1
* 注意：
* 1.自定义类必须是可序列化的
* 2.自定义类访问级别必须是Public
* 3.RDD转成DataFrame会把自定义类中字段的名称按assci码排序
*/
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("RDD");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile("sparksql/person.txt");
JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public Person call(String s) throws Exception {
Person p = new Person();
p.setId(s.split(",")[0]);
p.setName(s.split(",")[1]);
return p;
}
});
/**
* 传入进去Person.class的时候，sqlContext是通过反射的方式创建DataFrame
* 在底层通过反射的方式获得Person的所有field，结合RDD本身，就生成了DataFrame
*/
DataFrame df = sqlContext.createDataFrame(personRDD, Person.class); class Person implements Serializable {
private static final long serialVersionUID = -6907013906164009798L;
private String Id;
private String name; public void setId(String appId) {
this.appId = appId;
} public String getId() {
return appId;
} public String getname() {
return detail;
} public void setname(String detail) {
this.detail = detail;
}
}
//方法2：
JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt");
/**
* 转换成Row类型的RDD
*/
JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() { /**
*
*/
private static final long serialVersionUID = 1L; @Override
public Row call(String s) throws Exception {
return RowFactory.create(//这里字段顺序一定要和下边 StructField对应起来
String.valueOf(s.split(",")[0]),
String.valueOf(s.split(",")[1]),
);
}
});
/**
* 动态构建DataFrame中的元数据，一般来说这里的字段可以来源自字符串，也可以来源于外部数据库
*/
List<StructField> asList =Arrays.asList(//这里字段顺序一定要和上边对应起来
DataTypes.createStructField("id", DataTypes.StringType, true),
DataTypes.createStructField("name", DataTypes.StringType, true)
);
StructType schema = DataTypes.createStructType(asList);
/*
StructType schema = new StructType(new StructField[]{
new StructField("id", DataTypes.StringType, false, Metadata.empty()),
new StructField("name", DataTypes.StringType, false, Metadata.empty()),
});
*/
//DataFrame df = sqlContext.createDataFrame(List<Row> ,schema)这个方法也可以
DataFrame df = sqlContext.createDataFrame(rowRDD, schema);
//方法3
public static class Person implements Serializable {
private String name;
private int 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;
}
} // Create an instance of a Bean class
Person person = new Person();
person.setName("Andy");
person.setAge(32); // Encoders are created for Java beans
Encoder<Person> personEncoder = Encoders.bean(Person.class);
Dataset<Person> javaBeanDS = spark.createDataset(
Collections.singletonList(person),
personEncoder
);
javaBeanDS.show();
// +---+----+
// |age|name|
// +---+----+
// | 32|Andy|
// +---+----+ // Encoders for most common types are provided in class Encoders
Encoder<Integer> integerEncoder = Encoders.INT();
Dataset<Integer> primitiveDS = spark.createDataset(Arrays.asList(1, 2, 3), integerEncoder);
Dataset<Integer> transformedDS = primitiveDS.map(
(MapFunction<Integer, Integer>) value -> value + 1,
integerEncoder);
transformedDS.collect(); // Returns [2, 3, 4] // DataFrames can be converted to a Dataset by providing a class. Mapping based on name
String path = "examples/src/main/resources/people.json";
Dataset<Person> peopleDS = spark.read().json(path).as(personEncoder);
peopleDS.show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+