JSON数据集

• Scala
• Java
• Python
• R
• Sql

Spark SQL在加载JSON数据的时候，可以自动推导其schema并返回DataFrame。用SQLContext.read.json读取一个包含String的RDD或者JSON文件，即可实现这一转换。

注意，通常所说的json文件只是包含一些json数据的文件，而不是我们所需要的JSON格式文件。JSON格式文件必须每一行是一个独立、完整的的JSON对象。因此，一个常规的多行json文件经常会加载失败。

// sc是已有的SparkContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 数据集是由路径指定的
// 路径既可以是单个文件，也可以还是存储文本文件的目录
val path = "examples/src/main/resources/people.json"
val people = sqlContext.read.json(path)

// 推导出来的schema，可由printSchema打印出来
people.printSchema()
// root
//  |-- age: integer (nullable = true)
//  |-- name: string (nullable = true)

// 将DataFrame注册为table
people.registerTempTable("people")

// 跑SQL语句吧！
val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// 另一种方法是，用一个包含JSON字符串的RDD来创建DataFrame
val anotherPeopleRDD = sc.parallelize(
  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
val anotherPeople = sqlContext.read.json(anotherPeopleRDD)

Hive表

Spark SQL支持从Apache Hive读写数据。然而，Hive依赖项太多，所以没有把Hive包含在默认的Spark发布包里。要支持Hive，需要在编译spark的时候增加-Phive和-Phive-thriftserver标志。这样编译打包的时候将会把Hive也包含进来。注意，hive的jar包也必须出现在所有的worker节点上，访问Hive数据时候会用到（如：使用hive的序列化和反序列化SerDes时）。

Hive配置在conf/目录下hive-site.xml，core-site.xml（安全配置），hdfs-site.xml（HDFS配置）文件中。请注意，如果在YARN cluster（yarn-cluster mode）模式下执行一个查询的话，lib_mananged/jar/下面的datanucleus 的jar包，和conf/下的hive-site.xml必须在驱动器（driver）和所有执行器（executor）都可用。一种简便的方法是，通过spark-submit命令的–jars和–file选项来提交这些文件。

• Scala
• Java
• Python
• R

如果使用Hive，则必须构建一个HiveContext，HiveContext是派生于SQLContext的，添加了在Hive Metastore里查询表的支持，以及对HiveQL的支持。用户没有现有的Hive部署，也可以创建一个HiveContext。如果没有在hive-site.xml里配置，那么HiveContext将会自动在当前目录下创建一个metastore_db目录，再根据HiveConf设置创建一个warehouse目录（默认/user/hive/warehourse）。所以请注意，你必须把/user/hive/warehouse的写权限赋予启动spark应用程序的用户。

// sc是一个已有的SparkContext对象
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)

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

// 这里用的是HiveQL
sqlContext.sql("FROM src SELECT key, value").collect().foreach(println)

和不同版本的Hive Metastore交互

Spark SQL对Hive最重要的支持之一就是和Hive metastore进行交互，这使得Spark SQL可以访问Hive表的元数据。从Spark-1.4.0开始，Spark SQL有专门单独的二进制build版本，可以用来访问不同版本的Hive metastore，其配置表如下。注意，不管所访问的hive是什么版本，Spark SQL内部都是以Hive 1.2.1编译的，而且内部使用的Hive类也是基于这个版本（serdes，UDFs，UDAFs等）

以下选项可用来配置Hive版本以便访问其元数据：

用JDBC连接其他数据库

Spark SQL也可以用JDBC访问其他数据库。这一功能应该优先于使用JdbcRDD。因为它返回一个DataFrame，而DataFrame在Spark SQL中操作更简单，且更容易和来自其他数据源的数据进行交互关联。JDBC数据源在java和python中用起来也很简单，不需要用户提供额外的ClassTag。（注意，这与Spark SQL JDBC server不同，Spark SQL JDBC server允许其他应用执行Spark SQL查询）

首先，你需要在spark classpath中包含对应数据库的JDBC driver，下面这行包括了用于访问postgres的数据库driver

SPARK_CLASSPATH=postgresql-9.3-1102-jdbc41.jar bin/spark-shell

远程数据库的表可以通过Data Sources API，用DataFrame或者SparkSQL 临时表来装载。以下是选项列表：

• Scala
• Java
• Python
• R
• Sql

val jdbcDF = sqlContext.read.format("jdbc").options(
  Map("url" -> "jdbc:postgresql:dbserver",
  "dbtable" -> "schema.tablename")).load()

疑难解答

• JDBC driver class必须在所有client session或者executor上，对java的原生classloader可见。这是因为Java的DriverManager在打开一个连接之前，会做安全检查，并忽略所有对原声classloader不可见的driver。最简单的一种方法，就是在所有worker节点上修改compute_classpath.sh，并包含你所需的driver jar包。
• 一些数据库，如H2，会把所有的名字转大写。对于这些数据库，在Spark SQL中必须也使用大写。

性能调整

对于有一定计算量的Spark作业来说，可能的性能改进的方式，不是把数据缓存在内存里，就是调整一些开销较大的选项参数。

内存缓存

Spark SQL可以通过调用SQLContext.cacheTable(“tableName”)或者DataFrame.cache()把tables以列存储格式缓存到内存中。随后，Spark SQL将会扫描必要的列，并自动调整压缩比例，以减少内存占用和GC压力。你也可以用SQLContext.uncacheTable(“tableName”)来删除内存中的table。

你还可以使用SQLContext.setConf 或在SQL语句中运行SET key=value命令，来配置内存中的缓存。

其他配置选项

以下选项同样也可以用来给查询任务调性能。不过这些选项在未来可能被放弃，因为spark将支持越来越多的自动优化。

分布式SQL引擎

Spark SQL可以作为JDBC/ODBC或者命令行工具的分布式查询引擎。在这种模式下，终端用户或应用程序，无需写任何代码，就可以直接在Spark SQL中运行SQL查询。

运行Thrift JDBC/ODBC server

这里实现的Thrift JDBC/ODBC server和Hive-1.2.1中的HiveServer2是相同的。你可以使用beeline脚本来测试Spark或者Hive-1.2.1的JDBC server。

在Spark目录下运行下面这个命令，启动一个JDBC/ODBC server

./sbin/start-thriftserver.sh

这个脚本能接受所有 bin/spark-submit 命令支持的选项参数，外加一个 –hiveconf 选项，来指定Hive属性。运行./sbin/start-thriftserver.sh –help可以查看完整的选项列表。默认情况下，启动的server将会在localhost:10000端口上监听。要改变监听主机名或端口，可以用以下环境变量：

export HIVE_SERVER2_THRIFT_PORT=<listening-port>
export HIVE_SERVER2_THRIFT_BIND_HOST=<listening-host>
./sbin/start-thriftserver.sh \
  --master <master-uri> \
  ...

或者Hive系统属性 来指定

./sbin/start-thriftserver.sh \
  --hiveconf hive.server2.thrift.port=<listening-port> \
  --hiveconf hive.server2.thrift.bind.host=<listening-host> \
  --master <master-uri>
  ...

接下来，你就可以开始在beeline中测试这个Thrift JDBC/ODBC server:

./bin/beeline

下面的指令，可以连接到一个JDBC/ODBC server

beeline> !connect jdbc:hive2://localhost:10000

可能需要输入用户名和密码。在非安全模式下，只要输入你本机的用户名和一个空密码即可。对于安全模式，请参考beeline documentation.

Hive的配置是在conf/目录下的hive-site.xml，core-site.xml，hdfs-site.xml中指定的。

你也可以在beeline的脚本中指定。

Thrift JDBC server也支持通过HTTP传输Thrift RPC消息。以下配置（在conf/hive-site.xml中）将启用HTTP模式：

hive.server2.transport.mode - Set this to value: http
hive.server2.thrift.http.port - HTTP port number fo listen on; default is 10001
hive.server2.http.endpoint - HTTP endpoint; default is cliservice

同样，在beeline中也可以用HTTP模式连接JDBC/ODBC server:

beeline> !connect jdbc:hive2://<host>:<port>/<database>?hive.server2.transport.mode=http;hive.server2.thrift.http.path=<http_endpoint>

转载自 并发编程网 - ifeve.com