Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据

{"name":"ljs1","score":85}{"name":"ljs2","score":99}{"name":"ljs3","score":74}

package com.unicom.ljs.spark220.study;
import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;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.PairFunction;import org.apache.spark.sql.*;import org.apache.spark.sql.types.DataTypes;import org.apache.spark.sql.types.StructField;import org.apache.spark.sql.types.StructType;import scala.Tuple2;
import java.util.ArrayList;import java.util.List;
/** * @author: Created By lujisen * @company ChinaUnicom Software JiNan * @date: 2020-01-28 21:08 * @version: v1.0 * @description: com.unicom.ljs.spark220.study */public class JoinJsonData {    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("JoinJsonData");        JavaSparkContext sc=new JavaSparkContext(sparkConf);        SQLContext sqlContext=new SQLContext(sc);
        Dataset studentDS = sqlContext.read().json("D:\\dataML\\spark1\\student.json");        studentDS.registerTempTable("student_score");        Dataset studentNameScoreDS = sqlContext.sql("select  name,score  from student_score where  score > 82");
        List  studentNameList= studentNameScoreDS.javaRDD().map(new Function() {            @Override            public String call(Row row){                return row.getString(0);            }        }).collect();
        System.out.println(studentNameList.toString());
        List studentJsons=new ArrayList<>();        studentJsons.add("{\"name\":\"ljs1\",\"age\":18}");        studentJsons.add("{\"name\":\"ljs2\",\"age\":17}");        studentJsons.add("{\"name\":\"ljs3\",\"age\":19}");

        JavaRDD studentInfos = sc.parallelize(studentJsons);        Dataset studentNameScoreRDD = sqlContext.read().json(studentInfos);
        studentNameScoreRDD.schema();        studentNameScoreRDD.show();        studentNameScoreRDD.registerTempTable("student_age");
        String sql2="select  name,age  from student_age  where  name  in (";        for(int i=0;i            sql2+="'"+studentNameList.get(i)+"'";            if(i                sql2+=",";            }        }        sql2+=")";        Dataset studentNameAgeDS = sqlContext.sql(sql2);
        JavaPairRDD> studentNameScoreAge = studentNameScoreDS.toJavaRDD().mapToPair(new PairFunction() {            @Override            public Tuple2 call(Row row) throws Exception {                return new Tuple2(row.getString(0),                        Integer.valueOf(String.valueOf(row.getLong(1))));            }        }).join(studentNameAgeDS.toJavaRDD().mapToPair(new PairFunction() {            @Override            public Tuple2 call(Row row) throws Exception {                return new Tuple2(row.getString(0),                        Integer.valueOf(String.valueOf(row.getLong(1))));            }        }));        JavaRDD studentNameScoreAgeRow = studentNameScoreAge.map(new Function>, Row>() {            @Override            public Row call(Tuple2> v1) throws Exception {                return RowFactory.create(v1._1, v1._2._1, v1._2._2);            }        });
        List structFields=new ArrayList<>();        structFields.add(DataTypes.createStructField("name",DataTypes.StringType,true));        structFields.add(DataTypes.createStructField("score",DataTypes.IntegerType,true));        structFields.add(DataTypes.createStructField("age",DataTypes.IntegerType,true));

        StructType structType= DataTypes.createStructType(structFields);
        Dataset dataFrame = sqlContext.createDataFrame(studentNameScoreAgeRow, structType);
        dataFrame.schema();        dataFrame.show();        dataFrame.write().format("json").mode(SaveMode.Append).save("D:\\dataML\\spark1\\studentNameScoreAge");
    }}