Spark处理Dataset中数据

• 前言
• 一、转换算子
• • 1.1 flatMap函数
  • 1.2 map函数
  • 1.3 select和where
• 二、写出到文件
• • 1.1 写出格式为ORC
  • 1.2 写出到Parquet
  • 1.3 写出到Text

Dataset类，定义了很多算子转换函数。在这其中包含Map，flatMap、take、where、group等操作。

在Spark中算子执行的对象或是使用到对象，都需要序列化。否则会包类为找到和空指针异常
Encoders.bean(这里面的类不能是抽象类或是接口，且需实现Serializable接口

将当前数据一行转为多行。其中转换的数据类型是自定义的。

public Dataset datasetToFlatMap(Dataset ds){
	return ds.flatMap(new FlatMapFunction() {
     	@Override
     	public Iterator call(Row row) throws Exception {
        	List list = new ArrayList<>();
        	for(int i = 0 ; i < row.length(); i++){
            	list.add(row.getString(i));
         	}
         	return list.iterator();
     	}
	}, Encoders.bean(String.class));
}

public Dataset datasetToMap(Dataset ds){
	return ds.map(new MapFunction() {
		@Override
   		public String call(Row value) throws Exception {
       		return value.toString();
   		}
	},Encoders.bean(String.class));
}

需要注意的是select和where中的字段一定要在Dataset中存在
takeAsList(3) 返回当前数据样例前3条

public List datasetWhere(Dataset ds){
	return ds.select("v1","v2","v3")
	  		 .where("v3 = '"+v3+"'")
	  		 .takeAsList(3);
}

定义输出数据对象。对于已经定义了文件格式的需要注意：保存到文件的字段顺序是按字典顺序排列的。如果字段不是按字典排序定义的则需要修改参考下面代码的定义。

public class OrcHiveEntity implements Serializable {
    private static final long serialVersionUID = -11540L;
    //数据库字段位置： name, age, sex
    private String v1; //name
    private int v2;  //age
    private boolean sex; //sex

    public String getV1() {
        return v1;
    }

    public void setV1(String v1) {
        this.v1 = v1;
    }

    public int getV2() {
        return v2;
    }

    public void setV2(int v2) {
        this.v2 = v2;
    }

    public boolean isSex() {
        return sex;
    }

    public void setSex(boolean sex) {
        this.sex = sex;
    }

    @Override
    public String toString() {
        return "OrcHiveEntity{" +
                "v1='" + v1 + ''' +
                ", v2=" + v2 +
                ", sex=" + sex +
                '}';
    }
}

public static void writeOrcFile(Dataset dataset, String path){
	dataset.write().orc(path);
}

public static void writeOrcFile(Dataset dataset, String path){
	dataset.write().parquet(path);
}

需要追忆text文件，对象只能是一个字段的，不能是多个字段的。它是已一个字段为一行的

public static void writeOrcFile(Dataset dataset, String path){
	dataset.write().parquet(path);
}