import com.tencent.s2.util.KerberosAuthUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
import org.apache.spark.sql.functions.*;
import org.apache.spark.sql.expressions.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 测试spark代码的各种用法 java版本
 */
public class SparkJobTest {
    private static Logger logger = LoggerFactory.getLogger(SparkJobTest.class);
    public static void main(String[] args) {
        Calendar instance = Calendar.getInstance();
        instance.setTime(new Date());
        List<Student> students =new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            instance.add(Calendar.MONTH,i);
            String format = new SimpleDateFormat("yyyy-MM-dd").format(instance.getTime());
            Student student = new Student(i, "name" + i,format , 60 + i);
            students.add(student);
            System.out.println(student);;
        }
//        KerberosAuthUtil.kerberos_auth_cdp("hive@CDP.COM", "/data/DATA_DIR/share/keytab/hive.keytab");
        SparkJobTest jobTest = new SparkJobTest();
        SparkSession session = jobTest.buildSession();
        Dataset<Student> studentDataset = session.createDataset(students, Encoders.bean(Student.class));
        //缓存 多次使用
        Dataset<Student> studentDatasetCache = studentDataset.persist();
        Dataset<Row> renameColumn = studentDatasetCache.withColumnRenamed("birth", "birthDay");
        Dataset<Row> modifyColumn = studentDatasetCache.withColumn("score", new Column("score").plus(5));
        Dataset<Row> addColumn = studentDatasetCache.withColumn("sex", new Column("score").mod(2).equalTo(0));
        //union all 哪个在前面 字段名就用哪个的。
        modifyColumn.unionAll(renameColumn).show();
        //根据id name 去重，但是这个不能确定我要保留哪条数据 类似distinct
        renameColumn.unionAll(modifyColumn).dropDuplicates("id","name");

        Dataset<Row> exampleDataset = renameColumn
                .unionAll(modifyColumn).persist();
        //来个开窗函数
        exampleDataset
                .withColumn("rn",functions.row_number().over(Window.partitionBy("id").orderBy("score")))
                .select("*")
                .where(new Column("id").$greater$eq(1).and(new Column("rn").equalTo(1)))
                .orderBy("birthDay").show();
        //+----------+---+-----+-----+---+
        //|  birthDay| id| name|score| rn|
        //+----------+---+-----+-----+---+
        //|2022-12-10|  1|name1|   61|  1|
        //|2023-02-10|  2|name2|   62|  1|
        //|2023-05-10|  3|name3|   63|  1|
        //|2023-09-10|  4|name4|   64|  1|
        //|2024-02-10|  5|name5|   65|  1|
        //|2024-08-10|  6|name6|   66|  1|
        //|2025-03-10|  7|name7|   67|  1|
        //|2025-11-10|  8|name8|   68|  1|
        //|2026-08-10|  9|name9|   69|  1|
        //+----------+---+-----+-----+---+
        exampleDataset.groupBy(functions.substring(new Column("name"),0,4).as("name"))
                    .agg(functions.sum("score").alias("sumScore")
                            ,functions.sumDistinct("score").alias("sumDistinctScore")
                            ,functions.count("*").name("count*")
                            ,functions.countDistinct("id").as("countDistinctId")
                            )
                .select("*").show();
        //+----+--------+----------------+------+---------------+
        //|name|sumScore|sumDistinctScore|count*|countDistinctId|
        //+----+--------+----------------+------+---------------+
        //|name|    1340|            1005|    20|             10|
        //+----+--------+----------------+------+---------------+
        //注意exampleDataset row的内容是按照英文字母排序的。
        //一般来说我们期望的是id name birth 这样的顺序
        //但是exampleDataset .show()实际是 birth id  name 的顺序
        exampleDataset
                .select("id","name","birth")
                .write().insertInto("table");
        exampleDataset
                .select("id","name","birth")
                .write().partitionBy("birty").saveAsTable("table");
        session.stop();
    }

    private SparkSession buildSession() {
        SparkConf sparkConf = new SparkConf().setAppName("HouseRentSparkJob");
        if (onWindows()) {
            sparkConf.setMaster("local[*]");
            sparkConf.set("spark.driver.host", "localhost");
            sparkConf.set("spark.dirver.bindAddress", "127.0.0.1");
        }
        SparkSession session = SparkSession.builder()
                .config(sparkConf)
                .config("hive.exec.dynamic.partition.mode", "nonstrict")//动态分区
//                .config("spark.sql.hive.convertMetastoreOrc", "false")//解决读取不了hive元数据或者spark读取数据和hive不一致
                .config("spark.cleaner.referenceTracking.cleanCheckpoints", "true")//清理checkpoint
                .config("hive.metastore.dml.events", "false")//重复插入动态分区报错
                .enableHiveSupport()
                .getOrCreate();
        //一般情况不用checkpoint
        session.sparkContext().setCheckpointDir("/tmp/spark/job/OrderOnlineSparkJob");
        return session;
    }

    private boolean onWindows() {
        return System.getProperty("os.name").toLowerCase().contains("windows");
    }

}

import java.util.Date;

public class Student{
    public int id ;
    public String name;
    public String birth;
    public int score;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

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

    public String getBirth() {
        return birth;
    }

    public void setBirth(String birth) {
        this.birth = birth;
    }

    public int getScore() {
        return score;
    }

    public void setScore(int score) {
        this.score = score;
    }

    public Student(int id, String name, String birth, int score) {
        this.id = id;
        this.name = name;
        this.birth = birth;
        this.score = score;
    }

    public Student() {
    }

    @Override
    public String toString() {
        return "Student{" +
                "id=" + id +
                ", name='" + name + '\'' +
                ", birth=" + birth +
                ", score=" + score +
                '}';
    }
}

主要是几点。

1.通过functions.xxx去调用函数 涉及到 group  sum distinct  order by  desc ,,row_number()over(partition by )

2.insertInto 和saveAstable区别

insertInto是插入表，表不存在会报错

saveAstable是保存为表，每次会覆盖之前的表结构(有一定概率出错，出错了 hive删除表再跑就行)

字段问题 。比如我们建表 id name score

如果要insertinto 必须要select(id,name,score)

如果我们想saveastable 也建议select(id,name,score) ，不select不影响使用，只是字段位置按照英文字母顺序排列建的。 两者详见源码。。