Spark3教程六IDEA下Java开发Spark SQL

        上一篇文章中，我们使用了Scala语言调用Spark sql接口进行了开发，本篇文章我们使用Java语言进行同样业务功能的处理，依然是对JSON、Txt文本进行处理。
        JSON和Txt文件内容如下所示：

{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}

Michael, 29
Andy, 30
Justin, 19

        Java处理JSON代码：

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

public class Testsql {
    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder().master("local")
                .appName("Java Spark sql basic example")
                .config("spark.some.config.option", "some-value")
                .getorCreate();
        Dataset<Row> df = spark.read().json("file:///d:/test_spark/people.json");
        df.show();
        df.createOrReplaceTempView("people");
        Dataset<Row> sqlDF = spark.sql("select * from  people where age>20");
        sqlDF.show();
    }
}

        Java处理Txt代码，需要定义一个Person实体类：

public class Person {
    private String name;
    private long age;

    public String getName() {
        return name;
    }

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

    public long getAge() {
        return age;
    }

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

import com.alan.entity.Person;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.*;

public class TestText {
    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder().master("local")
                .appName("Java Spark sql basic example")
                .config("spark.some.config.option", "some-value")
                .getorCreate();
        JavaRDD<Person> peopleRDD = spark.read()
                .textFile("d:/test_spark/people.txt")
                .javaRDD()
                .map(line -> {
                    String[] parts = line.split(",");
                    Person person = new Person();
                    person.setName(parts[0]);
                    person.setAge(Integer.parseInt(parts[1].trim()));
                    return person;
                });
        // Apply a schema to an RDD of JavaBeans to get a DataFrame
        Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
        // Register the DataFrame as a temporary view
        peopleDF.createOrReplaceTempView("people");

        // sql statements can be run by using the sql methods provided by spark
        Dataset<Row> teenagersDF = spark.sql("select * from  people where age>20");
        teenagersDF.show();
    }
}