Spark SQL 整合 Hive 配置与使用

2. 如果 HDFS 配置了高可用，则还要把 Hadoop 集群中的 hdfs-site.xml 和 core-site.xml 文件拷贝到 Spark/conf 文件夹下面。

3. 启动 spark-shell 时指定 MySQL 连接驱动位置

Spark 集群模式:

bin/spark-shell \
  --master spark://spark-master:7077 \
  --executor-memory 512m \
  --total-executor-cores 2 \
  --driver-class-path /path/to/mysql-connector-java-5.1.35-bin.jar

Spark on YARN 模式:

bin/spark-shell \
  --master yarn \
  --executor-memory 512m \
  --total-executor-cores 2 \
  --driver-class-path /path/to/mysql-connector-java-5.1.35-bin.jar

4. 执行 SQL 语句

使用 SQLContext.sql 调用 HQL:

val rdd = sqlContext.sql("select * from default.person limit 2")
rdd.write.json("hdfs://192.168.19.131:9000/personresult")

使用 org.apache.spark.sql.hive.HiveContext:

import org.apache.spark.sql.hive.HiveContext
val hiveContext = new HiveContext(sc)
hiveContext.sql("select * from default.person")

5. 使用 spark-sql 命令启动 Shell 模式 启动 spark-sql 时指定 MySQL 连接驱动位置 (启动 spark-sql 那么就和 Hive 的操作一样，里面可以直接写 SQL 语句进行操作):

bin/spark-sql \
  --master spark://spark-master:7077 \
  --executor-memory 512m \
  --total-executor-cores 3 \
  --driver-class-path /path/to/mysql-connector-java-5.1.35-bin.jar

select * from default.person limit 2

Spark SQL 如何向元数据中添加数据？因为元数据库中只是存放表对应数据在 HDFS 的地址，并没有存放表的数据信息，Spark SQL 可以创建表，但是无法向表中添加数据比如 insert 语句。注意与把 DF 数据存储到数据库不是一个概念。

6. Thrift JDBC/ODBC Server Spark SQL 实现 Thrift JDBC/ODBC Server，这就意味着我们可以像 Hive 那样通过 JDBC 远程连接 Spark SQL 发送 SQL 语句并执行。在这之前需要先将 ${HIVE_HOME}/conf/hive-site.xml 拷贝到 ${SPARK_HOME}/conf 目录下，由于我的 Hive 配置了元数据信息存储在 MySQL 中，所以 Spark 在访问这些元数据信息时需要 MySQL 连接驱动的支持。

添加驱动的方式有三种： 第一种是在 ${SPARK_HOME}/conf 目录下的 spark-defaults.conf 中添加：spark.jars /path/to/mysql-connector-java-5.1.26-bin.jar。 第二种是通过添加：spark.driver.extraClassPath /path/to/mysql-connector-java-5.1.26-bin.jar 这种方式也可以实现添加多个依赖 jar，比较方便。 第三种是在运行时添加 --jars /path/to/mysql-connector-java-5.1.26-bin.jar。

启动 Thrift

在 Spark 根目录下执行：./sbin/start-thriftserver.sh 开启 Thrift 服务器。

./start-thriftserver.sh --jars /path/to/mysql-connector-java-5.1.35-bin.jar --master yarn

**start-thriftserver.sh 和 spark-submit 的用法类似，可以接受所有 spark-submit 的参数，并且还可以接受--hiveconf 参数。**不添加任何参数表示以 local 方式运行，默认的监听端口为 10000

用 Beeline 测试

在 Spark 根目录下执行：

./bin/beeline

连接 JDBC/ODBC Server

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

连接后会提示输入用户名和密码，用户名可以填当前登录的 Linux 用户名，密码为空即可。

在 Java 代码中用 JDBC 连接

接下来打开 Eclipse 用 JDBC 连接 HiveServer2，连接 Hive 的步骤同样如此。

在 pom.xml 添加以下依赖：

<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-jdbc</artifactId>
  <version>1.2.1</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
  <version>2.4.1</version>
</dependency>
<dependency>
  <groupId>jdk.tools</groupId>
  <artifactId>jdk.tools</artifactId>
  <version>1.6</version>
  <scope>system</scope>
  <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
</dependency>

驱动：org.apache.hive.jdbc.HiveDriver URL：jdbc:hive2://192.168.19.131:10000/default 用户名：hadoop (启动 ThriftServer 的 Linux 用户名) 密码："" (默认密码为空)

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
public class Test1 {
  public static void main(String[] args) throws SQLException {
    String url = "jdbc:hive2://192.168.19.131:10000/default";
    try {
      Class.forName("org.apache.hive.jdbc.HiveDriver");
    } catch (ClassNotFoundException e) {
      e.printStackTrace();
    }
    Connection conn = DriverManager.getConnection(url, "hadoop", "");
    Statement stmt = conn.createStatement();
    String sql = "SELECT * FROM person LIMIT 10";
    ResultSet res = stmt.executeQuery(sql);
    while(res.next()){
      System.out.println("id: "+res.getInt(1)+"	name: "+res.getString(2)+"	age:" + res.getInt(3));
    }
  }
}

这种方式，可以在 YARN 的管理界面看到，会长起一个任务，该任务负责跑 SQL 语句，但是不能并行跑 SQL 语句，就是同时为两个用户输入的查询语句同时跑，必须等一个跑完了再跑第二个。

Spark SQL 可视化

第一种方案： 将 Spark SQL 代码打包，SQL 语句和结果存储位置作为参数，Java 代码收集这些参数后，组装为命令，调用脚本向集群提交 JAR 包。 第二种方案： 根据 Spark 官网所述，Spark SQL 实现了 Thrift JDBC/ODBC Server。