13,JSON数据集(JSON Datasets)

    使用SQLContext.read.json()方法载入一个String RDD或者JSON文件

    需要注意的是,这里所谓的json文件不是传统的JSON文件.每一行必须包含一个分离的,自包含的标准的JSON对象.所以,一个多行的JSON文件无法读入.

    也就是说,这里载入的json文件如果有多行的话,每一行都必须是一个标准的JSON文件.

    如spark example自带的people.json文件:

     {"name":"Michael"}

     {"name":"Andy", "age":30}

     {"name":"Justin", "age":19}

14,Hive表(Hive Tables)

    Spark SQL同样支持从Hive中读取和写入数据.然而,由于Hive需要依赖很多其他的系统,在默认的Spark assembly中并不包含对Hive的支持.可以在编译Spark时添加参数-Phive和-Phive-thriftserver.

    初始化一个HiveContext对象,该对象继承自SQLContext,并且增加了在MetaStore中的表的查询功能和使用HiveQL查询的功能.及时没有安装Hive也可以使用HiveContext.如果没有hive-site.xml文件,则会自动的在当前文件夹中创建metastore_db和warehouse

    val sqlContext = new HiveContext(sc)

    与不同版本的Hive Metastore交互(Interacting with Different Versions of Hive Metastore)

15,使用JDBC连接其他的数据(JDBC To Other Databases)

    实现该功能需要使用到JdbcRDD类.

    程序运行时,需要包含JDBC驱动jar包.如果要在spark-shell中使用jdbc功能,需要按如下方式启动spark-shell:

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

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

各参数含义:

url	JDBC URL
dbtable	将要读入的表
driver	The class name of the JDBC driver needed to connect to this URL. This class will be loaded on the master and workers before running an JDBC commands to allow the driver to register itself with the JDBC subsystem.
partitionColumn, lowerBound, upperBound, numPartitions	These options must all be specified if any of them is specified. They describe how to partition the table when reading in parallel from multiple workers. partitionColumn must be a numeric column from the table in question. Notice that lowerBound and upperBound are just used to decide the partition stride, not for filtering the rows in table. So all rows in the table will be partitioned and returned.

(1)注意事项(Troubleshooting)

    a)JDBC驱动class必须在所有executors上可访问.这样实现的一个方法是修改所有worker节点上的compute_classpath.sh文件,使之包含这个driver JARs

    b)有些数据库,例如H2,会将所有names转化成大写,所以需要在Spark SQL中使用upper case来对应这些names

15,性能调优(Performance Tuning)

    (1)数据缓存到内存中(Caching Data In Memory)

    调用sqlContext.cacheTable("tableName")或者dataFrame.cache()将Spark SQL数据缓存到内存中.将内存中的表移除,sqlContext.uncacheTable("tableName")

    使用SQLContext对象的setConf方法或者使用SQL SET key=value命令

spark.sql.inMemoryColumnarStorage.compressed	true	Spark SQL会基于数据的信息自动为每一个字段选择压缩方式
spark.sql.inMemoryColumnarStorage.batchSize	1000	控制columnar 缓存的大小.参数越大效率越高,但是可能会导致内存溢出

    (2)其他可选配置(Other Configuration Options)