Spark2.2.0實戰(zhàn)中如何自動獲取Json文件元數(shù)據(jù)信息注冊兩個臨時表以及條件查詢后合并相同記錄數(shù)據(jù)

{"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<Row> studentDS = sqlContext.read().json("D:\\dataML\\spark1\\student.json");        studentDS.registerTempTable("student_score");        Dataset<Row> studentNameScoreDS = sqlContext.sql("select  name,score  from student_score where  score > 82");
        List<String>  studentNameList= studentNameScoreDS.javaRDD().map(new Function<Row, String>() {            @Override            public String call(Row row){                return row.getString(0);            }        }).collect();
        System.out.println(studentNameList.toString());
        List<String> studentJsons=new ArrayList<>();        studentJsons.add("{\"name\":\"ljs1\",\"age\":18}");        studentJsons.add("{\"name\":\"ljs2\",\"age\":17}");        studentJsons.add("{\"name\":\"ljs3\",\"age\":19}");

        JavaRDD<String> studentInfos = sc.parallelize(studentJsons);        Dataset<Row> 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<studentNameList.size();i++){            sql2+="'"+studentNameList.get(i)+"'";            if(i<studentNameList.size()-1){                sql2+=",";            }        }        sql2+=")";        Dataset<Row> studentNameAgeDS = sqlContext.sql(sql2);
        JavaPairRDD<String, Tuple2<Integer, Integer>> studentNameScoreAge = studentNameScoreDS.toJavaRDD().mapToPair(new PairFunction<Row, String, Integer>() {            @Override            public Tuple2<String, Integer> call(Row row) throws Exception {                return new Tuple2<String, Integer>(row.getString(0),                        Integer.valueOf(String.valueOf(row.getLong(1))));            }        }).join(studentNameAgeDS.toJavaRDD().mapToPair(new PairFunction<Row, String, Integer>() {            @Override            public Tuple2<String, Integer> call(Row row) throws Exception {                return new Tuple2<String, Integer>(row.getString(0),                        Integer.valueOf(String.valueOf(row.getLong(1))));            }        }));        JavaRDD<Row> studentNameScoreAgeRow = studentNameScoreAge.map(new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {            @Override            public Row call(Tuple2<String, Tuple2<Integer, Integer>> v1) throws Exception {                return RowFactory.create(v1._1, v1._2._1, v1._2._2);            }        });
        List<StructField> 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<Row> dataFrame = sqlContext.createDataFrame(studentNameScoreAgeRow, structType);
        dataFrame.schema();        dataFrame.show();        dataFrame.write().format("json").mode(SaveMode.Append).save("D:\\dataML\\spark1\\studentNameScoreAge");
    }}