import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;

/**
 * show how to use javardd api 
 */

public class HelloSparkJavaRDD {

    public final static String MASTER = "spark://master:7077";

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("helloSparkJava").setMaster(MASTER);
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaSparkContext.jarOfClass(HelloSparkJavaRDD.class);
        JavaRDD<String> lines = sc.textFile("README.md");
        //java8
        JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
            @Override
            public Integer call(String v1) throws Exception {
                return v1.length();
            }
        });
        int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        System.out.println("total length is :" + totalLength);
       /*  final String spark = "Spark";
        long numSpark= lines.filter(new Function<String, Boolean>() {
            public Boolean call(String v1) throws Exception {
                return v1.contains(spark);
            }
        }).count();
       System.out.println("find word '" + spark + "':" + numSpark);*/

    }
}

import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.Serializable;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

/**
 * Created by utf7 on 2016/9/19.
 * show how to user spark sql api
 */

public class HelloSparkJavaSQL {

    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder().appName("Java Spark SQL Example " + new Timestamp(System.currentTimeMillis())).config("spark.some.config.option", "some-value").getOrCreate();
        Dataset<Row> df = spark.read().json("file:/home/utf7/spark-2.0.0-bin-hadoop2.6/examples/src/main/resources/people.json");
        System.out.println("show df:");
        df.show();
        System.out.println("show printSchema:");
        df.printSchema();
        System.out.println("select name:");
        df.select("name").show();
        System.out.println("select name,age+1 :");
        df.select(new Column("name"), new Column("age").plus(1)).show();
        System.out.println("select age >21:");
        df.filter(new Column("age").gt(21)).show();
        System.out.println("select age,count(*) group by age :");
        df.groupBy("age").count().show();

        System.out.println("show how create view and select from the view:");
        df.createOrReplaceTempView("people");
        Dataset<Row> sqlDF = spark.sql("select * from people");
        sqlDF.show();

        System.out.println("show how to use dataset:");
        testDatasets(spark);
        testReadFromMySQL(spark);
    }


    /**
     * show how to read from rdms use jdbc,here is mysql
     * @param spark
     */
    public static void testReadFromMySQL(SparkSession spark) {
        String MYSQL_CONNECTION_URL = "jdbc:mysql://yc1:3306/test";
        String MYSQL_USERNAME = "utf7";
        String MYSQL_PWD = "utf7";
        String TABLE_NAME = "user";
        Person person = new Person();
        person.setName("utf7");
        person.setAge(28);
        // Encoders are created for Java beans

        Encoder<Person> personEncoder = Encoders.bean(Person.class);
        final Properties connectionProperties = new Properties();
        connectionProperties.put("user", MYSQL_USERNAME);
        connectionProperties.put("password", MYSQL_PWD);

        System.out.println("show how read from mysql:");
        Dataset<Person> peopleDS = spark.read().jdbc(MYSQL_CONNECTION_URL, TABLE_NAME, connectionProperties).as(personEncoder);
        peopleDS.show();

    }

    /**
     * show how to convert file to dataset,here is local file "file://..."
     * @param spark
     */
    public static void testDatasets(SparkSession spark) {
        Person person = new Person();
        person.setName("utf7");
        person.setAge(28);
        // 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(new MapFunction<Integer, Integer>() {
            @Override
            public Integer call(Integer value) throws Exception {
                return 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 = "file:/home/utf7/spark-2.0.0-bin-hadoop2.6/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|
        // +----+-------+

    }


    public static class Person implements Serializable {
        private String name;
        private int age;

        public String getName() {
            return this.name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public int getAge() {
            return age;
        }

        public void setAge(int age) {
            this.age = age;
        }
    }
}