生成dataset的幾種方式

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|
// +----+-------+