如何使用 Spark SQL

一、启动方法

/data/spark-1.4.0-bin-cdh4/bin/spark-sql --master spark://master:7077 --total-executor-cores 10 --executor-memory 1g --executor-cores 2

注：/data/spark-1.4.0-bin-cdh4/为spark的安装路径

/data/spark-1.4.0-bin-cdh4/bin/spark-sql –help 查看启动选项

--master MASTER_URL 指定master url

--executor-memory MEM 每个executor的内存，默认为1G

--total-executor-cores NUM 所有executor的总核数

-e <quoted-query-string> 直接执行查询SQL

-f <filename> 以文件方式批量执行SQL

二、Spark sql对hive支持的功能

1、查询语句：SELECT GROUP BY ORDER BY CLUSTER BY SORT BY

2、hive *** 作运算：

1) 关系运算：= ==, <>, <, >, >=, <=

2) 算术运算：+, -, *, /, %

3) 逻辑运算：AND, &&, OR, ||

4) 复杂的数据结构

5) 数学函数：(sign, ln, cos, etc)

6) 字符串函数：

3、 UDF

4、 UDAF

5、 用户定义的序列化格式

6、join *** 作：JOIN {LEFT|RIGHT|FULL} OUTER JOIN LEFT SEMI JOINCROSS JOIN

7、 unions *** 作：

8、 子查询： SELECT col FROM ( SELECT a + b AS col from t1) t2

9、Sampling

10、 Explain

11、 分区表

12、 视图

13、 hive ddl功能：CREATE TABLE、CREATE TABLE AS SELECT、ALTER TABLE

14、 支持的数据类型：TINYINT SMALLINT INT BIGINT BOOLEAN FLOAT DOUBLE STRING BINARY TIMESTAMPDATE ARRAY MAP STRUCT

三、Spark sql 在客户端编程方式进行查询数据

1、启动spark-shell

./spark-shell --master spark://master:7077 --total-executor-cores 10 --executor-memory 1g --executor-cores 2

2、编写程序

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

val df = sqlContext.read.json("../examples/src/main/resources/people.json")

查看所有数据：df.show()

查看表结构：df.printSchema()

只看name列：df.select("name").show()

对数据运算：df.select(df("name"), df("age") + 1).show()

过滤数据：df.filter(df("age") >21).show()

分组统计：df.groupBy("age").count().show()

1、查询txt数据

import sqlContext.implicits._

case class Person(name: String, age: Int)

val people = sc.textFile("../examples/src/main/resources/people.txt").map(_.split(",")).map(p =>Person(p(0), p(1).trim.toInt)).toDF()

people.registerTempTable("people")

val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")

2、parquet文件

val df = sqlContext.read.load("../examples/src/main/resources/users.parquet")

3、hdfs文件

val df = sqlContext.read.load("hdfs://namenode.Hadoop:9000/user/hive/warehouse/spark_test.db/test_parquet/part-r-00001.gz.parquet")

4、保存查询结果数据

val df = sqlContext.read.load("../examples/src/main/resources/users.parquet")

df.select("name", "favorite_color").write.save("namesAndFavColors.parquet“)

四、Spark sql性能调优

缓存数据表：sqlContext.cacheTable("tableName")

取消缓存表：sqlContext.uncacheTable("tableName")

spark.sql.inMemoryColumnarStorage.compressedtrue　当设置为true时，Spark SQL将为基于数据统计信息的每列自动选择一个压缩算法。

spark.sql.inMemoryColumnarStorage.batchSize　10000　柱状缓存的批数据大小。更大的批数据可以提高内存的利用率以及压缩效率，但有OOMs的风险

首先确保SQLSERVER服务正在运行，并可以允许远程访问连接

然后按如下步骤进行连接

1.安装JDBC驱动

1）下载JDBC驱动

2）执行sqljdbc_4.0.2206.100_chs.exe解压驱动文件

3）拷贝以下目录中的sqljdbc_auth.dll到Windows的System32目录。对x64的机子，x64的dll拷到C:\Windows\System32,x86的dll拷到C:\Windows\SysWOW64目录。

Microsoft JDBC Driver 4.0 for SQL Server\sqljdbc_4.0\chs\auth\

2.在Java程序中连接SQL Server

classpath中加上安装好的SQL Server jdbc jar包

Microsoft JDBC Driver 4.0 for SQL Server\sqljdbc_4.0\chs\sqljdbc4.jar

连接SQL Server的JDBC代码

Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver")

String url = "jdbc:sqlserver://localhost:54364databaseName=masterIntegratedSecurity=True"

Connection con = DriverManager.getConnection(url)

在SQL Server Browser服务开启的情况下也可以通过服务名连接

Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver")

String url = "jdbc:sqlserver://localhost\\SQLEXPRESSdatabaseName=masterIntegratedSecurity=True"

Connection con = DriverManager.getConnection(url)

3.身份验证模式

SQL Server默认使用Windows身份验证模式，这也是官方推荐的模式，安全性更高。上面的连接例子就是采用的Windows认证。如果要使用混合认证模式，需要下载SQL Server Management Studio,并通过SQL Server Management Studio修改认证模式为混合认证模式。

把Spark二进制包下载并解压到某一台*nux的机器上，这段代码中‘/Users/jilu/Downloads/’这段换成你自己的路径，这就是单

机执行SparkSQL的代码，在这个程序中，我已经创建好sqlContext了，以后的部分就是SparkSQL教程了。这是我更新完1.3版之后新

改的程序，不出意外1.X的版本都是这样用的。

PS：补充一下这个是Python API，不是Scala的。

import os

import sys

import traceback

# Path for spark source folder

os.environ['SPARK_HOME']="/Users/jilu/Downloads/spark-1.3.0-bin-hadoop2.4"

# Append pyspark to Python Path

sys.path.append("/Users/jilu/Downloads/spark-1.3.0-bin-hadoop2.4/python/")

sys.path.append("/Users/jilu/Downloads/spark-1.3.0-bin-hadoop2.4/python/lib/py4j-0.8.2.1-src.zip")

# try to import needed models

try:

from pyspark import SparkContext

from pyspark import SparkConf

from pyspark.sql import SQLContext, Row

print ("Successfully imported Spark Modules")

except ImportError as e:

print ("Can not import Spark Modules {}".format(traceback.format_exc()))

sys.exit(1)

# config spark env

conf = SparkConf().setAppName("myApp").setMaster("local")

sc = SparkContext(conf=conf)

sqlContext = SQLContext(sc)

---