189 8069 5689

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文件元数据信息注册两个临时表以及条件查询后合并相同记录数据
本文URL:http://ptruijie.cn/article/pdcjej.html

联系我们

您好HELLO!
感谢您来到新都网站建设公司,若您有合作意向,请您为我们留言或使用以下方式联系我们, 我们将尽快给你回复,并为您提供真诚的设计服务,谢谢。
  • 电话:028- 86922220 18980695689
  • 商务合作邮箱:631063699@qq.com
  • 合作QQ: 532337155
  • 成都网站设计地址:成都市青羊区锣锅巷31号五金站写字楼6楼

广皓图文建站工作室

新都广皓图文网站建设公司拥有多年以上互联网从业经验的团队,始终保持务实的风格,以"帮助客户成功"为已任,专注于提供对客户有价值的服务。 我们已为众企业及上市公司提供专业的网站建设服务。我们不只是一家网站建设的网络公司;我们对营销、技术、管理都有自己独特见解,广皓图文建站采取“创意+综合+营销”一体化的方式为您提供更专业的服务!

广皓观点

相对传统的新都网站建设公司而言,广皓图文是互联网中的网站品牌策划,我们精于企业品牌与互联网相结合的整体战略服务。
我们始终认为,网站必须注入企业基因,真正使网站成为企业vi的一部分,让整个网站品牌策划体系变的深入而持久。