文档章节

查询分数大于80分的学生信息java版本实例

别寒
 别寒
发布于 2017/07/27 11:40
字数 630
阅读 23
收藏 0
package cn.hhb.spark.sql;

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.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataType;
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;

/**
 * Created by dell on 2017/7/27.
 */
public class JSONDataSource {
    public static void main(String[] args) {

        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("JSONDataSource").setMaster("local")
                .set("spark.testing.memory", "2147480000");

        // 创建javasparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        // 针对json文件,创建dataframe(针对json文件创建dataframe)
        DataFrame studentScoresDF = sqlContext.read().json("c://students.json");

        // 针对学生成绩信息的dataframe,注册临时表,查询分数大于80的学生
        // (注册临时表,针对临时表执行sql语句)
        studentScoresDF.registerTempTable("student_scores");
        DataFrame goodStudentScoresDF = sqlContext.sql("select name,score from student_scores where score >= 80");

        // (将dataframe转换为rdd,执行transframation操作)
        List<String> goodStudentNames = goodStudentScoresDF.javaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                return row.getString(0);
            }
        }).collect();

        // 然后针对javaRDD<String>,创建DataFrame
        // (针对包含json串的javardd,创建dataframe)
        List<String> studentInfoJSONs = new ArrayList<String>();
        studentInfoJSONs.add("{\"name\":\"Leo\", \"age\":18}");
        studentInfoJSONs.add("{\"name\":\"Marry\", \"age\":17}");
        studentInfoJSONs.add("{\"name\":\"Jack\", \"age\":19}");
        JavaRDD<String> studentInfoJSONsRDD = sc.parallelize(studentInfoJSONs);
        DataFrame studentInfosDF = sqlContext.read().json(studentInfoJSONsRDD);

        // 针对学生基本信息DataFrame,注册临时表,然后查询分数大于80分的学生的基本信息
        studentInfosDF.registerTempTable("student_infos");
        String sql = "select name,age from student_infos where name in(";
        for (int i=0; i<goodStudentNames.size(); i++) {
            sql += "'" + goodStudentNames.get(i) + "'";
            if (i < goodStudentNames.size() - 1){
                sql += ",";
            }
        }
        sql += ")";

        DataFrame goodStudentInfosDF = sqlContext.sql(sql);

        // 然后将两份数据的dataframe,转换为javaPairRDD,执行join transframation
        // (将dataframe转换为javardd,再map为JavaPairRDD,然后进行join)
        JavaPairRDD<String, Tuple2<Integer, Integer>> goodStudentsRDD =
                goodStudentScoresDF.javaRDD().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(goodStudentInfosDF.javaRDD().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)))
                );
            }
        }));

        // 将封装在rdd中的好学生的全部信息,转换为一个javardd<row>的格式
        // (将javardd转换为dataframe)
        // 就是之前 以编程方式动态指定元数据,将rdd转换为dataframe 的知识点
        JavaRDD<Row> goodStudentRowsRDD = goodStudentsRDD.map(new Function<Tuple2<String, Tuple2<Integer, Integer>>, Row>() {
            @Override
            public Row call(Tuple2<String, Tuple2<Integer, Integer>> tuple) throws Exception {
                return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
            }
        });

        // 然后动态构造一份元数据,将javardd<row>转换为dataframe
        List<StructField> structFields = new ArrayList<StructField>();
        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);

        // 使用动态构造的元数据,将rdd转换为dataframe
        DataFrame goodStudentsDF = sqlContext.createDataFrame(goodStudentRowsRDD, structType);

        // 将dataframe转换为javardd,然后打印
        List<Row> rows = goodStudentsDF.javaRDD().collect();
        for (Row row : rows){
            System.out.println(row);
        }

        // 将好学生的全部信息保存到一个json文件中去
        // (将dataframe中的数据保存到json文件中去)
        goodStudentsDF.write().format("json").save("c://goodStudent.json");

        sc.close();

    }
}

© 著作权归作者所有

共有 人打赏支持
别寒
粉丝 30
博文 271
码字总数 137605
作品 0
永州
程序员
私信 提问
mybtis 等标签的作用

sql语句where条件中,需要一些安全判断,例如按性别检索,如果传入的参数是空的,此时查询出的结果很可能是空的,也许我们需要参数为空 时,是查出全部的信息。这是我们可以使用动态sql,增加...

北京_
2015/03/09
0
0
Criteria 和 DetachedCriteria 的主要区别《摘录》

Criteria 和 DetachedCriteria 的主要区别在于创建的形式不一样, Criteria 是在线的,所 以它是由 Hibernate Session 进行创建的;而 DetachedCriteria 是离线的,创建时无需 Session,Det...

西行侠客
2014/04/10
0
0
安装logstash及logstash的初步使用-处理DNS日志

安装logstash 查询java信息 rpm -qa | grep java 查询yum源内含有的所有java的安装包 yum -y list java* 选择需要的版本 yum install java-xxx-openjdk.x86_64 安装完成后,验证版本 java -...

进击的大佬
08/23
0
0
Elasticsearch入门实践

一. 系统环境 操作系统:CentOS release 6.8 (Final) ES版本:6.1.1 二. 安装 先确认安装了Java运行时环境: 解压ES压缩包: 三. 启动 1. 启动ES单节点 当然,对于在后台以守护进程模式运行的...

哲别0
06/06
0
0
Linux系统梳理---系统搭建(一):jdk卸载与安装

1.去官网下载符合Linux版本的jdk,暂用jdk-8u171-linux-x64.rpm 2.登陆Linux,进入usr目录,创建java目录(方便管理,可以其他位置):mkdir java 3.上传下载的jdk包至Linux服务器,使用rz指令(sz f...

勤奋的蚂蚁
07/16
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Go 使用channel控制并发

前言 channel一般用于协程之间的通信,channel也可以用于并发控制。比如主协程启动N个子协程,主协程等待所有子协程退出后再继续后续流程,这种场景下channel也可轻易实现。 场景示例 总结 ...

恋恋美食
24分钟前
1
0
Apache Flink 漫谈系列 - 持续查询(Continuous Queries)

摘要: 实际问题 我们知道在流计算场景中,数据是源源不断的流入的,数据流永远不会结束,那么计算就永远不会结束,如果计算永远不会结束的话,那么计算结果何时输出呢?本篇将介绍Apache Fl...

阿里云官方博客
27分钟前
4
0
斐波那契堆的理解,节点mark属性和势函数

斐波那契堆 看了好多博客,都是照搬算法导论的内容,没有自己的理解,比如为什么有mark属性,势函数的作用,以及为什么叫斐波那契堆,下面说说鄙人的理解。 势函数 势函数是根节点个数加上2...

杨喆
29分钟前
3
0
NIO源码详解

阻塞io和无阻塞io: 阻塞io是指jdk1.4之前版本面向流的io,服务端需要对每个请求建立一堆线程等待请求,而客户端发送请求后,先咨询服务端是否有线程相应,如果没有则会一直等待或者遭到拒 ...

沉稳2018
33分钟前
0
0
如何把已经提交的commit, 从一个分支放到另一个分支

在本地master提交了一个commit(8d85d4bca680a5dbcc3e5cfb3096d18cd510cc9f),如何提交的test_2分之上? git checkout test_2git cherry-pick 8d85d4bca680a5dbcc3e5cfb3096d18cd510cc9f......

stephen_wu
37分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部