平武建站

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

这篇文章将为大家详细讲解有关Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

创新互联科技有限公司专业互联网基础服务商，为您提供服务器机柜租用，高防服务器，成都IDC机房托管，成都主机托管等互联网服务。

   Spark支持两个方式将RDD转换成DataFrame

    1.反射;将schema信息定义在一个单独的class中，通过这个scheme转换成对应的DataFrame,这种方式简单，但不建议用，因为scala的case class最多只支持22个字段，所以必须要自己开发一个类，实现product接口。

    2.通过编程接口，自己构建StruntType,将RDD转换成对应的DataFrame,这种方式稍微麻烦，官网手册列出大体三个步骤：

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

    翻译一下大体意思：

    1.创建RDD转换成JavaRDD

    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 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");
    }}

关于Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。


标题名称：Spark2.2.0实战中如何自动获取Json文件元数据信息注册两个临时表以及条件查询后合并相同记录数据
当前网址：http://pwwzsj.com/article/pdcjej.html

其他资讯