第1关：SparkSQL初识

package com.educoder.bigData.sparksql;

import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;


public class Test1 {
	

	public static void main(String[] args) throws AnalysisException {
		/********* Begin *********/
		SparkSession  spark  =  SparkSession   
                  .builder()  
                  .appName("Java Spark SQL基本示例")  
                  .master("local")  
                  .config("spark.some.config.option" , "some-value")  
                  .getOrCreate();  
 //打印spark版本号  
       System.out.println(spark.version());  
		
		
		
		
		
		
		/********* End *********/
	}

	

}

第2关：Dataset创建及使用

package com.educoder.bigData.sparksql;

import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.SparkSession;


public class Test2 {


	public static void main(String[] args) throws AnalysisException {
		
		SparkSession  spark  =  SparkSession 
				  .builder()
				  .appName("test1")
				  .master("local")
				  .config("spark.some.config.option" , "some-value")
				  .getOrCreate();
		/********* Begin *********/
		spark.read().json("people.json").createOrReplaceTempView("people");  
spark.sql("select * from people where age != '23'").show();
		
		
		
		
		
		/********* End *********/
	}

	

}

第3关：Dataset自定义函数

package com.educoder.bigData.sparksql;
import java.util.ArrayList;  
import java.util.List;
import org.apache.spark.sql.Row;  
import org.apache.spark.sql.expressions.MutableAggregationBuffer;  
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;  
import org.apache.spark.sql.types.DataType;  
import org.apache.spark.sql.types.DataTypes;  
import org.apache.spark.sql.types.StructField;  
import org.apache.spark.sql.types.StructType;  
public class MyAverage extends UserDefinedAggregateFunction {  
    private static final long serialVersionUID = 1L;  
    private StructType inputSchema;  
    private StructType bufferSchema;  
    public MyAverage() {  
        List<StructField> inputFields = new ArrayList<StructField>();  
        inputFields.add(DataTypes.createStructField("inputColumn", DataTypes.LongType, true));  
        inputSchema = DataTypes.createStructType(inputFields);  
        List<StructField> bufferFields = new ArrayList<StructField>();  
        bufferFields.add(DataTypes.createStructField("sum", DataTypes.LongType, true));  
        bufferFields.add(DataTypes.createStructField("count", DataTypes.LongType, true));  
        bufferSchema = DataTypes.createStructType(bufferFields);  
      }  
    @Override  
    public StructType bufferSchema() {  
        // TODO Auto-generated method stub  
        return bufferSchema;  
    }  
    @Override  
    public DataType dataType() {  
        // TODO Auto-generated method stub  
        return DataTypes.DoubleType;  
    }  
    @Override  
    public boolean deterministic() {  
        // TODO Auto-generated method stub  
        return true;  
    }  
    @Override  
    public Object evaluate(Row buffer) {  
        // TODO Auto-generated method stub  
         return ((double) buffer.getLong(0)) / buffer.getLong(1);  
    }  
    @Override  
    public void initialize(MutableAggregationBuffer buffer) {  
        // TODO Auto-generated method stub  
        buffer.update(0, 0L);  
        buffer.update(1, 0L);  
    }  
    @Override  
    public StructType inputSchema() {  
        // TODO Auto-generated method stub  
        return inputSchema;  
    }  
    @Override  
    public void merge(MutableAggregationBuffer buffer1, Row buffer2) {  
        // TODO Auto-generated method stub  
        long mergedSum = buffer1.getLong(0) + buffer2.getLong(0);  
        long mergedCount = buffer1.getLong(1) + buffer2.getLong(1);  
        buffer1.update(0, mergedSum);  
        buffer1.update(1, mergedCount);  
    }  
    @Override  
    public void update(MutableAggregationBuffer buffer, Row input) {  
        if (!input.isNullAt(0)) {  
            long updatedSum = buffer.getLong(0) + input.getLong(0);  
            long updatedCount = buffer.getLong(1) + 1;  
            buffer.update(0, updatedSum);  
            buffer.update(1, updatedCount);  
        }  
    }  
}