文档章节

spark sql简单示例

cloud-coder
 cloud-coder
发布于 2015/06/16 17:36
字数 437
阅读 19828
收藏 8

运行环境

集群环境:CDH5.3.0

具体JAR版本如下:

spark版本:1.2.0-cdh5.3.0

hive版本:0.13.1-cdh5.3.0

hadoop版本:2.5.0-cdh5.3.0

spark sql的JAVA版简单示例

  1. spark sql直接查询JSON格式的数据

  2. spark sql的自定义函数

  3. spark sql查询hive上面的表

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.SparkConf;
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.sql.api.java.DataType;
import org.apache.spark.sql.api.java.JavaSQLContext;
import org.apache.spark.sql.api.java.JavaSchemaRDD;
import org.apache.spark.sql.api.java.Row;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.hive.api.java.JavaHiveContext;


/**
 * 注意:
 * 使用JavaHiveContext时
 * 1:需要在classpath下面增加三个配置文件:hive-site.xml,core-site.xml,hdfs-site.xml
 * 2:需要增加postgresql或mysql驱动包的依赖
 * 3:需要增加hive-jdbc,hive-exec的依赖
 *
 */
public class SimpleDemo {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("simpledemo").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaSQLContext sqlCtx = new JavaSQLContext(sc);
        JavaHiveContext hiveCtx = new JavaHiveContext(sc);
//        testQueryJson(sqlCtx);
//        testUDF(sc, sqlCtx);
        testHive(hiveCtx);
        sc.stop();
        sc.close();
    }

    //测试spark sql直接查询JSON格式的数据
    public static void testQueryJson(JavaSQLContext sqlCtx) {
        JavaSchemaRDD rdd = sqlCtx.jsonFile("file:///D:/tmp/tmp/json.txt");
        rdd.printSchema();

        // Register the input schema RDD
        rdd.registerTempTable("account");

        JavaSchemaRDD accs = sqlCtx.sql("SELECT address, email,id,name FROM account ORDER BY id LIMIT 10");
        List<Row> result = accs.collect();
        for (Row row : result) {
            System.out.println(row.getString(0) + "," + row.getString(1) + "," + row.getInt(2) + ","
                    + row.getString(3));
        }

        JavaRDD<String> names = accs.map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                return row.getString(3);
            }
        });
        System.out.println(names.collect());
    }


    //测试spark sql的自定义函数
    public static void testUDF(JavaSparkContext sc, JavaSQLContext sqlCtx) {
        // Create a account and turn it into a Schema RDD
        ArrayList<AccountBean> accList = new ArrayList<AccountBean>();
        accList.add(new AccountBean(1, "lily", "lily@163.com", "gz tianhe"));
        JavaRDD<AccountBean> accRDD = sc.parallelize(accList);

        JavaSchemaRDD rdd = sqlCtx.applySchema(accRDD, AccountBean.class);

        rdd.registerTempTable("acc");

        // 编写自定义函数UDF
        sqlCtx.registerFunction("strlength", new UDF1<String, Integer>() {
            @Override
            public Integer call(String str) throws Exception {
                return str.length();
            }
        }, DataType.IntegerType);

        // 数据查询
        List<Row> result = sqlCtx.sql("SELECT strlength('name'),name,address FROM acc LIMIT 10").collect();
        for (Row row : result) {
            System.out.println(row.getInt(0) + "," + row.getString(1) + "," + row.getString(2));
        }
    }

    //测试spark sql查询hive上面的表
    public static void testHive(JavaHiveContext hiveCtx) {
        List<Row> result = hiveCtx.sql("SELECT foo,bar,name from pokes2 limit 10").collect();
        for (Row row : result) {
            System.out.println(row.getString(0) + "," + row.getString(1) + "," + row.getString(2));
        }
    }
}


© 著作权归作者所有

共有 人打赏支持
cloud-coder
粉丝 249
博文 191
码字总数 135000
作品 0
广州
架构师
私信 提问
加载中

评论(2)

i
iawom
D-_-D
D-_-D
JavaSQLContext 是哪个maven 依赖下的哦,都没找到呢
【Spark】Spark SQL, DataFrames and Datasets Guide(翻译文,持续更新)

本文主要是翻译Spark官网Spark SQL programming guide 。只能保证大概意思,尽量保证细节。英文水平有限,如果有错误的地方请指正,轻喷。目录导航在右上角 Spark SQL、DataFrames 和 Datase...

跑呀跑
09/19
0
0
hive,shark,sparkSQL,hive on spark,impala,drill比较

Hive on Mapreduce Hive的原理大家可以参考这篇大数据时代的技术hive:hive介绍,实际的一些操作可以看这篇笔记:新手的Hive指南,至于还有兴趣看Hive优化方法可以看看我总结的这篇Hive性能优...

hblt-j
08/13
0
0
如何使用Spark SQL 的JDBC server

简介 Spark SQL provides JDBC connectivity, which is useful for connecting business intelligence (BI) tools to a Spark cluster and for sharing a cluster across multipleusers. The......

cloud-coder
2015/06/17
0
0
3.sparkSQL整合Hive

  spark SQL经常需要访问Hive metastore,Spark SQL可以通过Hive metastore获取Hive表的元数据。从Spark 1.4.0开始,Spark SQL只需简单的配置,就支持各版本Hive metastore的访问。注意,涉...

intsmaze(刘洋)
08/09
0
0
Spark 学习资源收集【Updating】

(一)spark 相关安装部署、开发环境 1、Spark 伪分布式 & 全分布式 安装指南 http://my.oschina.net/leejun2005/blog/394928 2、Apache Spark探秘:三种分布式部署方式比较 http://dongxic...

大数据之路
2014/09/08
0
1

没有更多内容

加载失败,请刷新页面

加载更多

大数据教程(7.5)hadoop中内置rpc框架的使用教程

博主上一篇博客分享了hadoop客户端java API的使用,本章节带领小伙伴们一起来体验下hadoop的内置rpc框架。首先,由于hadoop的内置rpc框架的设计目的是为了内部的组件提供rpc访问的功能,并不...

em_aaron
11分钟前
0
0
CentOS7+git+github创建Python开发环境

1.准备CentOS7 (1)下载VMware Workstation https://pan.baidu.com/s/1miFU8mk (2)下载CentOS7镜像 https://mirrors.aliyun.com/centos/ (3)安装CentOS7系统 http://blog.51cto.com/fengyuns......

枫叶云
昨天
1
0
利用ibeetl 实现selectpicker 的三级联动

1. js 直接写在html页面上面,ibeetl 就可以动态地利用后台传上来的model List ,不需要每次点击都要ajax请求后台 2. 使用selectpicker 的时候,除了对selecct option的动态处理后,还需要 $("#...

donald121
昨天
1
0
Android SELinux avc dennied权限问题解决方法

1. 概述 SELinux是Google从android 5.0开始,强制引入的一套非常严格的权限管理机制,主要用于增强系统的安全性。 然而,在开发中,我们经常会遇到由于SELinux造成的各种权限不足,即使拥有“...

TreasureWe
昨天
3
0
阿里云ACP认证详细笔记(一)

ECS--------------------------1.云服务器Elastic Compute Service(ECS)2.Terraform:您可以使用开源工具Terraform来预配和管理ECS资源。Terraform提供一种简单机制,能够将配置文件部署...

啃不动地大坚果
昨天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部