Spark Sql

一、Spark SQL

1. sqlContext

  • SparkContext 创建上下文
  • 基于 SparkContext 创建 sqlContext 上下文
  • sqlContext 操作源数据

1.1 测试数据源

  1. vim /data/log/customers.txt
  3. 100, John Smith, Austin, TX, 78727
  4. 200, Joe Johnson, Dallas, TX, 75201
  5. 300, Bob Jones, Houston, TX, 77028
  6. 400, Andy Davis, San Antonio, TX, 78227
  7. 500, James Williams, Austin, TX, 78727

1.2 sqlContext 编程 - 指定模式 操作 sql

  1. import org.apache.spark.{SparkConf, SparkContext};
  3. val sparkConf = new SparkConf()
  4.     sparkConf.setAppName("SparkSQLHiveOnYarn")
  5.     sparkConf.setMaster("local[2]")
  6. val sc = new SparkContext(sparkConf)
  8. val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  10. // 创建RDD对象
  11. val rddCustomers = sc.textFile("/data/log/customers.txt")
  13. // 导入 Spark SQL Row
  14. import org.apache.spark.sql.Row;
  16. // 导入 Spark SQL 数据类型
  17. import org.apache.spark.sql.types.{StructType, StructField, StringType};
  19.  // 用字符串编码模式
  20. val schemaString = "customer_id name city state zip_code"
  21. // 用模式字符串生成模式对象
  22. val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
  24. // 将RDD（rddCustomers）记录转化成Row。
  25. val rowRDD = rddCustomers.map(_.split(",")).map(p => Row(p(0).trim,p(1),p(2),p(3),p(4)))
  27. // 将模式应用于RDD对象。
  28. val dfCustomers = sqlContext.createDataFrame(rowRDD, schema)
  30. // 将DataFrame注册为表
  31. dfCustomers.registerTempTable("customers")
  33. // 用sqlContext对象提供的sql方法执行SQL语句。
  34. val custNames = sqlContext.sql("SELECT name FROM customers")
  36. // SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。
  37. // 可以按照顺序访问结果行的各个列。
  38. custNames.map(t => "Name: " + t(0)).collect().foreach(println)
  40. // 用sqlContext对象提供的sql方法执行SQL语句。
  41. val customersByCity = sqlContext.sql("SELECT name,zip_code FROM customers ORDER BY zip_code")
  43. // SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。
  44. // 可以按照顺序访问结果行的各个列。
  45. customersByCity.map(t => t(0) + "," + t(1)).collect().foreach(println)

2. SparkContext -> HiveContext 操作 hive table

  2. import org.apache.spark.{SparkConf, SparkContext};
  3. import org.apache.spark.sql._;
  4. import org.apache.spark.sql.hive.HiveContext;
  6. // spark 设置配置与上线文
  7. val sparkConf = new SparkConf()
  8. sparkConf.setAppName("SparkSQLHiveOnYarn")
  9. sparkConf.setMaster("local[2]")
  10. val sc = new SparkContext(sparkConf)
  12. // 基于 SparkContext 上下文配置 HiveContext
  13. val hiveContext = new HiveContext(sc)
  14. // 配置 hive 元数据服务
  15. hiveContext.setConf("hive.metastore.uris", "thrift://NameNode:9083")
  17. // 通过 hiveContext 操作 hive table 数据, 写法一
  18.     hiveContext.sql("SHOW DATABASES").collect().foreach { x => println(x) }
  19.     hiveContext.sql("SHOW TABLES").collect().foreach { x => println(x) }
  21.     hiveContext.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
  22.     hiveContext.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")
  23.     // Queries are expressed in HiveQL
  24.     hiveContext.sql("FROM src SELECT key, value").collect().foreach(println)
  26. // 通过 hiveContext 操作 hive 的数据, 写法二
  27.     // 导入语句，可以隐式地将 RDD 转化成 DataFrame
  28.     import hiveContext.implicits._  ;
  29.     import hiveContext.sql;
  30.     sql("SHOW DATABASES").collect().foreach { x => println(x) }
  31.     sql("SHOW TABLES").collect().foreach { x => println(x) }

3. 通过 Thriftserver 连接 Hive

  • 启动 spark 的 Thriftserver
  • 通过 hive 客户端连接 Thriftserver 服务器
  • 操作 hive 数据
  2. import java.util.Map;
  3. import java.util.HashMap;
  4. import java.util.List;
  5. import java.util.ArrayList;
  6. import java.util.Arrays;
  8. import java.sql.SQLException;
  9. import java.sql.Connection;
  10. import java.sql.ResultSet;
  11. import java.sql.Statement;
  12. import java.sql.DriverManager;
  13. import org.apache.hive.jdbc.HiveDriver;
  15. public class HiveClient {
  17.     private static String driverName = "org.apache.hive.jdbc.HiveDriver";
  19.     /**
  20.      * 获取连接
  21.      */
  22.     private Connection connection;
  23.     public Connection getConnection() {
  24.         return connection;
  25.     }
  26.     public void setConnection(Connection conn) {
  27.         this.connection = conn;
  28.     }
  31.     /**
  32.      * 获取连接句柄
  33.      */
  34.     private Statement stmt;
  35.     public Statement getStmt() {
  36.         return stmt;
  37.     }
  38.     public void setStmt(Statement stmt) {
  39.         this.stmt = stmt;
  40.     }
  43.     public HiveClient(String url, String user, String password) throws SQLException {
  44.         // 导入类
  45.         try {
  46.             Class.forName(driverName);
  47.         } catch (ClassNotFoundException e) {
  48.             e.printStackTrace();
  49.         }
  51.         // 创建连接
  52.         Connection con = DriverManager.getConnection(url, user, password);
  53.         this.setConnection(con);
  55.         // 创建连接句柄语句
  56.         Statement stmt = con.createStatement();
  57.         this.setStmt(stmt);
  58.     }
  61.     /**
  62.      * 执行指定 Sql
  63.      * @param sql
  64.      * @return 布尔值
  65.      */
  66.     public Boolean execute (String sql) {
  67.         Boolean rs = false;
  68.         try {
  69.             rs = this.getStmt().execute(sql);
  70.         } catch (SQLException e) {
  71.             e.printStackTrace();
  72.         }
  73.         return rs;
  74.     }
  77.     /**
  78.      * 查询数据
  79.      * @param sql
  80.      * @param fields 字段
  81.      * @return List<Map<String, String>>
  82.      * @throws SQLException
  83.      */
  84.     public List<Map<String, String>> select(String sql, String fields) throws SQLException {
  85.         // 保存结果数据
  86.         List<Map<String, String>> listResult = new ArrayList<Map<String, String>>();
  88.         ResultSet res = null;
  89.         try {
  90.             res = this.getStmt().executeQuery(sql);
  91.         } catch (SQLException e) {
  92.             e.printStackTrace();
  93.         }
  95.         // 转换为数组
  96.         String[] arrFields = fields.split(",");
  98.         // 遍历每一行
  99.         while (res.next()) {
  100.             // 保存一行数据
  101.             Map<String, String> mapRowData = new HashMap<String, String>();
  103.             // 拼接字段值
  104.             for (String field : arrFields) {
  105.                 mapRowData.put(field, res.getString("broker_id"));
  106.             }
  107.             // 追加到 list 中
  108.             listResult.add(mapRowData);
  109.         }
  111.         /**
  112.         for (Map<String, String> rs : listResult) {
  113.             System.out.println(rs.get("broker_id"));
  114.         }
  115.         */
  117.         return listResult;
  118.     }
  121.     /**
  122.      * 关闭连接
  123.      */
  124.     public void closeConnection() {
  125.         try {
  126.             this.getConnection().close();
  127.         } catch (SQLException e) {
  128.             e.printStackTrace();
  129.         }
  130.     }
  132. }