Spark1.0新特性-->Spark SQL

Spark1.0出来了,变化还是挺大的,文档比以前齐全了,RDD支持的操作比以前多了一些,Spark on
yarn功能我居然跑通了。但是最最重要的就是多了一个Spark
SQL的功能,它能对RDD进行Sql操作,目前它只是一个alpha版本,喜欢尝鲜的同志们进来看看吧,下面是它的官网的翻译。

Spark
SQL是支持在Spark中使用Sql、HiveSql、Scaca中的关系型查询表达式。它的核心组件是一个新增的RDD类型SchemaRDD,它把行对象用一个Schema来描述行里面的所有列的数据类型,它就像是关系型数据库里面的一张表。它可以从原有的RDD创建,也可以是Parquet文件,最重要的是它可以支持用HiveQL从hive里面读取数据。

下面是一些案例,可以在Spark shell当中运行。

首先我们要创建一个熟悉的Context,熟悉spark的人都知道吧,有了Context我们才可以进行各种操作。

val sc: SparkContext // 已经存在的SparkContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

import sqlContext._

Running SQL on RDDs


Spark SQL支持的一种表的类型是Scala的case class,case class定义了表的类型,下面是例子:


val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._

// case class在Scala 2.10里面最多支持22个列,,为了突破这个现实,最好是定义一个类实现Product接口
case class Person(name: String, age: Int)

// 为Person的对象创建一个RDD,然后注册成一张表
val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt))
people.registerAsTable("people")

// 直接写sql吧,这个方法是sqlContext提供的
val teenagers = sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// teenagers是SchemaRDDs类型,它支持所有普通的RDD操作
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

从上面这个方法来看,不是很好用,一个表好几十个字段,我就得一个一个的去赋值,它现在支持的操作都是很简单的操作,想要实现复杂的操作可以具体去看HiveContext提供的HiveQL。

Using Parquet


val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._

val people: RDD[Person] = ... // 同上面的例子.

// 这个RDD已经隐式转换成一个SchemaRDD, 允许它存储成Parquet格式.
people.saveAsParquetFile("people.parquet")

// 从上面创建的文件里面读取,加载一个Parquet文件的结果也是一种JavaSchemaRDD.
val parquetFile = sqlContext.parquetFile("people.parquet")

//注册成表,然后使用
parquetFile.registerAsTable("parquetFile")
val teenagers = sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.collect().foreach(println)

Writing
Language-Integrated Relational Queries


目前这个功能只是在Scala里面支持,挺鸡肋的一个功能

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._
val people: RDD[Person] = ... // 同前面的例子.

// 和后面这个语句是一样的 ‘SELECT name FROM people WHERE age >= 10 AND age <= 19‘
val teenagers = people.where(‘age >= 10).where(‘age <= 19).select(‘name)

Hive Support


这下面的才是高潮,它可以从hive里面取数据。但是hive的依赖太多了,默认Spark
assembly是没带这些依赖的,需要我们运行SPARK_HIVE=true sbt/sbt
assembly/assembly重新编译,或者用maven的时候添加-Phive参数,它会重新编译出来一个hive assembly的jar包,然后需要把这个jar包放到所有的节点上。另外还需要把hive-site.xml放到conf目录下。没进行hive部署的话,下面的例子也可以用LocalHiveContext来代替HiveContext。


val sc: SparkContext // 已经存在的SparkContext
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)

// 引入这个Context,然后就会给所有的sql语句进行隐式转换
import hiveContext._

hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
hql("LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt‘ INTO TABLE src")

// 使用HiveQL查询
hql("FROM src SELECT key, value").collect().foreach(println)

这个功能看起来还挺像样,前面两个看起来就像渣一样,没劲儿,不知道为什么不自带那些依赖,还要我们再编译一下,但是我下的那个版本运行的时候提示我已经编译包括了hive的。尼玛,真恶心。

时间: 2024-10-05 03:25:27

Spark1.0新特性-->Spark SQL的相关文章

Spark1.0.0新特性

Spark1.0.0 release于2014-05-30日正式公布,标志Spark正式进入1.X的时代.Spark1.0.0带来了各种新的特性,并提供了更好的API支持:Spark1.0.0添加了Spark SQL这一个新的重要组件,用于载入和操作Spark的结构化数据:Spark1.0.0增强了现有的标准库(ML,streaming,GraphX),同一时候还增强了Java和Python语言的支持:最后,Spark1.0.0在运维上做了非常大的改进,包含支持Hadoop/YARN安全机制.使

Spark整合kafka0.10.0新特性(二)

接着Spark整合kafka0.10.0新特性(一)开始 import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.common.serialization.StringDeserializer import org.apache.spark.streaming.kafka010._ import org.apache.spark.streaming.kafka010.LocationStrat

Entity Framework Core 2.0 新特性

本文翻译来自:https://docs.microsoft.com/en-us/ef/core/what-is-new/index 一.模型级查询过滤器(Model-level query filters) ef core2.0包含了一个新特性,我们叫他模型级查询过滤器(Model-level query filters).此特性允许使用Linq查询表达式直接定义在实体类型的元数据模型上.这样的过滤器会自动应用到任何LINQ查询所涉及的那些实体类型,包括间接引用的实体类型(对象引用,导航属性).

[转]Servlet 3.0 新特性详解

原文地址:http://blog.csdn.net/xiazdong/article/details/7208316 Servlet 3.0 新特性概览 1.Servlet.Filter.Listener无需在web.xml中进行配置,可以通过Annotation进行配置: 2.模块化编程,即将各个Servlet模块化,将配置文件也分开配置. 3.Servlet异步处理,应对复杂业务处理: 4.异步Listener,对于异步处理的创建.完成等进行监听: 5. 文件上传API简化: tomcat

Java基础加强-(注解,类加载器,servlet3.0新特性)

1.   Annotation注解 1.1.  Annotation概述 Annotation是JDK 5.0以后提供对元数据的支持,可以在编译.加载和运行时被读取,并执行相应的处理.所谓Annotation就是提供了一种为程序元素设置元数据的方法,可用于修饰包.类.构造器.方法.成员变量.参数和局部变量的声明,这些信息被存储在Annotation的“name=value”对中. Annotation能被用来为程序元素(类.方法.成员变量等)设置元数据,比如一段代码的作者或者告诉编译器禁止一些特

Mysql 8.0 新特性测试

Mysql 8.0 新特性测试 Role MySQL8.0版本添加了role特性,role是一种逻辑概念是权限的集合,可以将一个或以上的权限赋予给role,再将role赋给user.Oracle,Postgresql和Mariadb中早已存在role这个特性. create role role_test; grant select,insert,delete,update on zhongwc.tab01 to role_test; create user 'user1'@'%' identif

MySQL8.0新特性【转】

Server层,选项持久化 mysql> show variables like '%max_connections%'; +------------------------+-------+ | Variable_name | Value | +------------------------+-------+ | max_connections | 512 | | mysqlx_max_connections | 100 | +------------------------+-------

Day07 jdk5.0新特性&Junit&反射

day07总结 今日内容 MyEclipse安装与使用 JUnit使用 泛型 1.5新特性 自动装箱拆箱 增强for 静态导入 可变参数方法 枚举 反射 MyEclipse安装与使用(yes) 安装MyEclipse 先安装了JDK ? MyEclipse介绍 ? MyEclipse是Eclipse的一个插件: MyEclipse是需要花钱的: MyEclipse官网不在欢迎中国人登录: ? MyEclipse使用 ? 1 创建项目 选择工作空间: 工作空间路径不能有空格和中文: 工作空间以班名

Atitit.&#160;C#.net&#160;clr&#160;2.0&#160;&#160;4.0新特性

Atitit. C#.net clr 2.0  4.0新特性 1. CLR内部结构1 2. CLR 版本发展史3 3. CLR 2.0 3 4. CLR 4 新特性 概览4 4.1.1.  托管与本地代码的互操作5 4.1.2.    垃圾回收6 4.1.3.    代码约定6 4.1.4.    Corrupted state exception6 4.1.5.     新的安全模型7 4.1.6.     同一个进程,多个CLR7 4.1.7.     基本类库7 5. CLR最新发展8 6