这篇文章将为大家详细讲解有关Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。
Spark支持两个方式将RDD转换成DataFrame
1.反射;将schema信息定义在一个单独的class中，通过这个scheme转换成对应的DataFrame,这种方式简单，但不建议用，因为scala的case class最多只支持22个字段，所以必须要自己开发一个类，实现product接口。
2.通过编程接口，自己构建StruntType,将RDD转换成对应的DataFrame,这种方式稍微麻烦，官网手册列出大体三个步骤：
翻译一下大体意思：
1.创建RDD转换成JavaRDD<Row>
2.按照Row的数据结构定义StructType
3.基于StructType使用createDataFrame创建DataFrame
数据准备：
第一个json文件student.json
{"name":"ljs1","score":85}{"name":"ljs2","score":99}{"name":"ljs3","score":74}
第二个json数据，直接写在了代码的低46-49行中，可直接查看代码获取
代码实例：
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>() {@Overridepublic 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>() {@Overridepublic 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>() {@Overridepublic 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>() {@Overridepublic 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");}}
关于Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
