问题对人有帮助，内容完整，我也想知道答案
问题没有实际价值，缺少关键内容，没有改进余地
spark sql中运行报错
SELECT his.name, his.oid FROM his_data_zadd AS his WHERE his.value=(SELECT MAX(temp_t.value) FROM his_data_zadd AS temp_t)
报错信息：
py4j.protocol.Py4JJavaError: An error occurred while calling o32.sql.
: java.lang.RuntimeException: [1.76] failure: ``)'' expected but identifier MAX found
SELECT his.name, his.oid FROM his_data_zadd AS his WHERE his.value=(SELECT MAX(temp_t.value) FROM his_data_zadd AS temp_t)
是spark sql的语法解析器不支持这种写法？还是我的sql语句原本就有问题？
答案对人有帮助，有参考价值
答案没帮助，是错误的答案，答非所问
试下把SQLContext换成HiveContext
同步到新浪微博
分享到微博？
你好！看起来你挺喜欢这个内容，但是你还没有注册帐号。 当你创建了帐号，我们能准确地追踪你关注的问题，在有新答案或内容的时候收到网页和邮件通知。还能直接向作者咨询更多细节。如果上面的内容有帮助，记得点赞 (????)? 表示感谢。
明天提醒我
关闭理由：
删除理由：
忽略理由：
推广（招聘、广告、SEO 等）方面的内容
与已有问题重复（请编辑该提问指向已有相同问题）
答非所问，不符合答题要求
宜作评论而非答案
带有人身攻击、辱骂、仇恨等违反条款的内容
无法获得确切结果的问题
非开发直接相关的问题
非技术提问的讨论型问题
其他原因（请补充说明）
我要该，理由是：
扫扫下载 App1.in 不支持子查询 eg. select * from src where key in(select key from test);
支持查询个数 eg. select * from src where key in(1,2,3,4,5);
in 40000个 耗时25.766秒
in 80000个 耗时78.827秒
2.union all/union 不支持顶层的union all eg. select key from src UNION ALL
支持select * from (select key from src union all select key from test) 不支持 union 支持select distinct key from (select key from src union all select key from test) 3.intersect 不支持
4.minus 不支持
5.except 不支持
6.inner join/join/left outer join/right outer join/full outer join/left semi join 都支持
left outer join/right outer join/full outer join 中间必须有outer join是最简单的关联操作，两边关联只取交集; left outer join是以左表驱动，右表不存在的key均赋值为null； right outer join是以右表驱动，左表不存在的key均赋值为null； full outer join全表关联，将两表完整的进行笛卡尔积操作，左右表均可赋值为 left semi join最主要的使用场景就是解决 Hive不支持where子句中的子查询，SQL常用的exist in子句在Hive中是不支持的 不支持子查询 eg. select * from src aa where aa.key in(select bb.key from test bb); 可用以下两种方式替换： select * from src aa left outer join test bb on aa.key=bb.key where bb.key && select * from src aa left semi join test bb on aa.key=bb. 大多数情况下 JOIN ON 和 left semi on 是对等的 A,B两表连接，如果B表存在重复数据 当使用JOIN ON的时候，A,B表会关联出两条记录，应为ON上的条件符合；
而是用LEFT SEMI JOIN 当A表中的记录，在B表上产生符合条件之后就返回，不会再继续查找B表记录了， 所以如果B表有重复，也不会产生重复的多条记录。
left outer join 支持子查询 eg. select aa.* from src aa left outer join (select * from test111)bb on aa.key=bb.a; 7. hive四中数据导入方式 1）从本地文件系统中导入数据到Hive表
create table wyp(id int,name string) ROW FORMAT delimited fields terminated by '\t' STORED AS TEXTFILE;
load data local inpath 'wyp.txt' 2)从HDFS上导入数据到Hive表
[wyp@master /home/q/hadoop-2.2.0]$ bin/hadoop fs -cat /home/wyp/add.txt
hive& load data inpath '/home/wyp/add.txt' 3)从别的表中查询出相应的数据并导入到Hive表中
hive& create table test(
& id int, name string
& ,tel string)
& partitioned by
& (age int)
& ROW FORMAT DELIMITED
& FIELDS TERMINATED BY '\t'
& STORED AS TEXTFILE;
注：test表里面用age作为了分区字段，分区：在Hive中，表的每一个分区对应表下的相应目录，所有分区的数据都是存储在对应的目录中。
比如wyp表有dt和city两个分区，则对应dt=city=BJ对应表的目录为/user/hive/warehouse/dt=/city=BJ，
所有属于这个分区的数据都存放在这个目录中。
hive& insert into table test
& partition (age='25')
& select id, name, tel
也可以在select语句里面通过使用分区值来动态指明分区：
hive& set hive.exec.dynamic.partition.mode=
hive& insert into table test
& partition (age)
& select id, name,
& tel, age
Hive也支持insert overwrite方式来插入数据
hive& insert overwrite table test
& PARTITION (age)
& select id, name, tel, age
Hive还支持多表插入
hive& from wyp
& insert into table test
& partition(age)
& select id, name, tel, age
& insert into table test3
& select id, name
& where age&25;
4)在创建表的时候通过从别的表中查询出相应的记录并插入到所创建的表中
hive& create table test4
& select id, name, tel
& 8.查看建表语句 hive& show create table test3;
9.表重命名 hive& ALTER TABLE events RENAME TO 3
10.表增加列 hive& ALTER TABLE pokes ADD COLUMNS (new_col INT);
11.添加一列并增加列字段注释
hive& ALTER TABLE invites ADD COLUMNS (new_col2 INT COMMENT 'a comment');
12.删除表 hive& DROP TABLE
hive& select * from test order by key limit 10;14.创建数据库 Create D
14.alter table tablename &change&oldColumn newColumn column_type 修改列的名称和类型
alter table yangsy CHANGE product_no phone_no string
15.导入.sql文件中的sql
&spark-sql&--driver-class-path&/home/hadoop/hive/lib/mysql-connector-java-5.1.30-bin.jar&-f&testsql.sql&
insert into table CI_CUSER_51522 select mainResult.PRODUCT_NO,dw_coclbl_m02__02_01,dw_coclbl_d01__01_04 from (select
PRODUCT_NO from CI_CUSER_32267) mainResult
left join DW_COCLBL_M02_201407 dw_coclbl_m02_3848 on mainResult.PRODUCT_NO = dw_coclbl_m02_3848.PRODUCT_NO left join DW_COCLBL_D01_ dw_coclbl_d01_3845 on dw_coclbl_m02_3848.PRODUCT_NO = dw_coclbl_d01_3845.PRODUCT_NO
into CI_CUSER_23638 ( PRODUCT_NO,ATTR_COL_0000,ATTR_COL_0001) select mainResult.PRODUCT_NO,dw_coclbl_m02__02_01,dw_coclbl_m02__03_01 from (select
PRODUCT_NO from CI_CUSER_32267) mainResult
left join DW_COCLBL_M02_201407 dw_coclbl_m02_3848 on mainResult.PRODUCT_NO = dw_coclbl_m02_3848.PRODUCT_NO
TABLE ci_cuser_yymmddhhmisstttttt_tmp(product_no string) row format serde 'com.bizo.hive.serde.csv.CSVSerde' ;
LOAD DATA LOCAL INPATH '/home/ocdc/coc/yuli/test123.csv' OVERWRITE INTO TABLE test_yuli2;
创建支持CSV格式的testfile文件 CREATE
TABLE test_yuli7 row format serde 'com.bizo.hive.serde.csv.CSVSerde' as select * from CI_CUSER_29786;
不依赖CSVSerde的jar包创建逗号分隔的表"create table " +listName+ " ROW FORMAT DELIMITED FIELDS TERMINATED BY ','" +" as select * from " + listName1;
create table aaaa ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' STORED AS TEXTFILE as select * from
ThriftServer 开启FAIR模式SparkSQL Thrift Server 开启FAIR调度方式:1. 修改$SPARK_HOME/conf/spark-defaults.conf,新增2. spark.scheduler.mode
FAIR3. spark.scheduler.allocation.file
/Users/tianyi/github/community/apache-spark/conf/fair-scheduler.xml4. 修改$SPARK_HOME/conf/fair-scheduler.xml(或新增该文件), 编辑如下格式内容5. &?xml version="1.0"?&6. &allocations&7. &pool name="production"&8.
&schedulingMode&FAIR&/schedulingMode&9.
&!-- weight表示两个队列在minShare相同的情况下,可以使用资源的比例 --&10.
&weight&1&/weight&11.
&!-- minShare表示优先保证的资源数 --&12.
&minShare&2&/minShare&13. &/pool&14. &pool name="test"&15.
&schedulingMode&FIFO&/schedulingMode&16.
&weight&2&/weight&17.
&minShare&3&/minShare&18. &/pool&19. &/allocations&20. 重启Thrift Server21. 执行SQL前,执行 22. set spark.sql.thriftserver.scheduler.pool=指定的队列名
等操作完了 create table yangsy555 like CI_CUSER_YYMMDDHHMISSTTTTTT
然后insert into yangsy555 select * from yangsy555
创建一个自增序列表，使用row_number() over()为表增加序列号 以供分页查询
create&table&yagnsytest2&as&SELECT&ROW_NUMBER()&OVER()&as&id,*&from&
Sparksql的解析与Hiveql的解析的执行流程:
阅读(...) 评论()Spark SQL是支持在Spark中使用Sql、HiveSql、Scala中的关系型查询表达式。它的核心组件是一个新增的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._
Data Sources（数据源）
Spark SQL通过SchemaRDD接口支持在多种数据源上进行操作。一旦一个数据集被加载，它可以被注册成一个表格，甚至可以和其它数据源有连接。
&Spark SQL支持的一种表的类型是Scala的case class，case class定义了表的类型，下面是例子：
// sc是一个已经存在的SprakContextval sqlContext = new org.apache.spark.sql.SQLContext(sc)
// import sqlContext._
import sqlContext.createSchemaRDD
// 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。
Parquet Files
Parquet是一种列式存储格式并且被许多数据处理系统支持。Parquet为Hadoop生态系统中的所有项目提供支持高效率压缩的列式数据表达，而且与数据处理框架、数据模型或编程语言都没有关系。Spark SQL提供了对Parquet的读和写，自动保留原始数据的架构。
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// import sqlContext._
// createSchemaRDD被用来将RDD隐式转换成一个SchemaRDDimport sqlContext.createSchemaRDD
val people: RDD[Person] = ... // 同上面的例子.
// 这个RDD已经隐式转换成一个SchemaRDD, 允许它存储成Parquet格式.
people.saveAsParquetFile("people.parquet")
// 从上面创建的文件里面读取，加载一个Parquet文件的结果也是一种JavaSchemaRDD.
val parquetFile = sqlContext.parquetFile("people.parquet")
//注册成表，然后在SQL状态下使用
parquetFile.registerAsTable("parquetFile")
val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age &= 13 AND age &= 19")
teenagers.map(t =& "Name:" + t(0)).collect().foreach(println)
JSON Datasets(JSON数据集）
JSON(JavaScript Object Notation) 是一种轻量级的数据交换格式。它基于JavaScript（Standard ECMA-262 3rd Edition - December 1999）的一个子集。 JSON采用完全独立于语言的文本格式，但是也使用了类似于C语言家族的习惯（包括C, C++, C#, Java, JavaScript, Perl, Python等）。这些特性使JSON成为理想的数据交换语言。易于人阅读和编写，同时也易于机器解析和生成(网络传输速度快)。
SparkSQL可以自动推断出一个JSON数据集模式并作为一个SchemaRDD来加载。这种转换可以通过使用SQLContext中的两个方法中的一个得到：
jsonFile - 从JSON文件的目录中加载数据，其中文件的每一行就是一个JSON对象。
jsonRdd - 从一个已存在的RDD中加载数据，其中每一个RDD元素都是一个包含一个JSON对象的字符串。
// sc 是已经存在的SparkContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// 一个JSON 数据集用一个路径指出
// 这个路径既可以是一个单独的文本文件，也可以是一个存储文本文件的目录
val path = "examples/src/main/resources/people.json"
// 根据路径指出的文件生成一个SchemaRDD
val people = sqlContext.jsonFile(path)
// 推断的模式可以通过使用printSchema() 方法显式化
people.printSchema()
|-- age: IntegerType
|-- name: StringType
// 把SchemaRDD注册成一个表
people.registerAsTable("people")
// SQL状态可以通过使用sqlContext提供的sql方法运行
val teenagers = sqlContext.sql("SELECT name FROM people WHERE age &= 13 AND age &= 19")
// 另外，一个SchemaRDD也可以通过每个字符串存储一个JSON数据集对象的string类型的RDD来生成
val anotherPeopleRDD = sc.parallelize(
"""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
val anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)
Hive Tables
&Spark SQL也支持读写存储在Apache 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)
或者写成如下形式：
// sc is an existing SparkContext.
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
hiveContext.hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
hiveContext.hql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")
// Queries are expressed in HiveQL
hiveContext.hql("FROM src SELECT key, value").collect().foreach(println)
Writing Language-Integrated Relational Queries
文字语言综合关联查询，目前这个功能只是在Scala里面支持。
Spark SQL还支持一个特定域的语言编写查询。再次，利用上述实例数据：
// sc是一个已经存在的SparkContextval 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)
teenagers.map(t =& "Name: " + t(0)).collect().foreach(println)
DSL（领域语言）使用scala符号表示隐含表中的列，通过在前面加一个（&）来标示。隐式转换将这些符号表达式表示成SQL执行引擎的值。一个完整的功能支持列表可以在ScalaDoc中找到。
阅读(...) 评论()sparksql的基本使用
sparksql的基本使用。sparksql是简化，是使用sql来分析数据，跟sql的用法很相似，可以直接在命令行里面使用sql语句进行分析，查询，同时也可以使用代码来进行编程，下面我来简单的介绍下使用scala语言进行sparksql的编程：里面涉及到了RDD和Dataframe和DataSet之间的区别。
package org.apache.spark.examples.sql
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.Encoder
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
object SparkSQLExample {
//定义匹配类
case class Person(name: String, age: Long)
def main(args: Array[String]) {
//创建sparkSeesion
val spark = SparkSession
.builder()
.appName(&Spark SQL basic example&)
.config(&spark.some.config.option&, &some-value&)
.master(&local[*]&)
.getOrCreate()
//导入隐式转换
import spark.implicits._
//下面是各种方法，具体的可以看下面的具体的代码
runBasicDataFrameExample(spark)
runDatasetCreationExample(spark)
runInferSchemaExample(spark)
runProgrammaticSchemaExample(spark)
spark.stop()
*基本的使用
private def runBasicDataFrameExample(spark: SparkSession): Unit = {
// 读取一个json数据格式的问价
val df = spark.read.json(&examples/src/main/resources/people.json&)
//输出数据
// +----+-------+
// +----+-------+
// |null|Michael|
19| Justin|
// +----+-------+
//导入隐式转换
import spark.implicits._
//输出json文件的约束
df.printSchema()
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)
// 只显示name列
df.select(&name&).show()
// +-------+
// +-------+
// |Michael|
// | Justin|
// +-------+
// 查找name 和age+1
df.select($&name&, $&age& + 1).show()
// +-------+---------+
name|(age + 1)|
// +-------+---------+
// |Michael|
// | Justin|
// +-------+---------+
// 筛选出年龄大于21的
df.filter($&age& & 21).show()
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+
// 用年龄进行分组
df.groupBy(&age&).count().show()
// +----+-----+
// | age|count|
// +----+-----+
// +----+-----+
//创建一个临时的people表
df.createOrReplaceTempView(&people&)
//执行sql语句
val sqlDF = spark.sql(&SELECT * FROM people&)
sqlDF.show()
// +----+-------+
// +----+-------+
// |null|Michael|
19| Justin|
// +----+-------+
*DataSet的使用
private def runDatasetCreationExample(spark: SparkSession): Unit = {
import spark.implicits._
//创建一个people然后转化为DataSet
val caseClassDS = Seq(Person(&Andy&, 32)).toDS()
caseClassDS.show()
// +----+---+
// |name|age|
// +----+---+
// |Andy| 32|
// +----+---+
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)
val path = &examples/src/main/resources/people.json&
val peopleDS = spark.read.json(path).as[Person]
peopleDS.show()
// +----+-------+
// +----+-------+
// |null|Michael|
19| Justin|
// +----+-------+
*带有约束的表（带有字段的表）
private def runInferSchemaExample(spark: SparkSession): Unit = {
import spark.implicits._
// 创建一个RDD 然后转化成dataframe
val peopleDF = spark.sparkContext
.textFile(&examples/src/main/resources/people.txt&)
.map(_.split(&,&))
.map(attributes =& Person(attributes(0), attributes(1).trim.toInt))
//创建一个临时的people表
peopleDF.createOrReplaceTempView(&people&)
val teenagersDF = spark.sql(&SELECT name, age FROM people WHERE age BETWEEN 13 AND 19&)
//通过下标查数据
teenagersDF.map(teenager =& &Name: & + teenager(0)).show()
// +------------+
// +------------+
// |Name: Justin|
// +------------+
// 通过field name查数据
teenagersDF.map(teenager =& &Name: & + teenager.getAs[String](&name&)).show()
// +------------+
// +------------+
// |Name: Justin|
// +------------+
//这点据说还很重要，但是不是太理解
// No pre-defined encoders for Dataset[Map[K,V]], define explicitly
implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]
//也可以写成下面这个样子
// implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()
// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
teenagersDF.map(teenager =& teenager.getValuesMap[Any](List(&name&, &age&))).collect()
// Array(Map(&name& -& &Justin&, &age& -& 19))
*RDD和约束的结合
private def runProgrammaticSchemaExample(spark: SparkSession): Unit = {
import spark.implicits._
//创建一个 RDD
val peopleRDD = spark.sparkContext.textFile(&examples/src/main/resources/people.txt&)
// The schema is encoded in a string
val schemaString = &name age&
// 把一个字符串数组转化成一个fields
val fields = schemaString.split(& &)
.map(fieldName =& StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)
// Convert records of the RDD (people) to Rows
val rowRDD = peopleRDD
.map(_.split(&,&))
.map(attributes =& Row(attributes(0), attributes(1).trim))
//把这个约束应用到这个RDD上
val peopleDF = spark.createDataFrame(rowRDD, schema)
peopleDF.createOrReplaceTempView(&people&)
val results = spark.sql(&SELECT name FROM people&)
results.map(attributes =& &Name: & + attributes(0)).show()
// +-------------+
// +-------------+
// |Name: Michael|
Name: Andy|
// | Name: Justin|
// +-------------+