Spark SQL实战(08)-整合Hive_spark整合hive

1 整合原理及使用

Apache Spark 是一个快速、可扩展的分布式计算引擎，而 Hive 则是一个数据仓库工具，它提供了数据存储和查询功能。在 Spark 中使用 Hive 可以提高数据处理和查询的效率。

场景

历史原因积累下来的，很多数据原先是采用Hive来进行处理的，现想改用Spark操作数据，须要求Spark能够无缝对接已有的Hive的数据，实现平滑过渡。

MetaStore
Hive底层的元数据信息是存储在MySQL中，$HIVE_HOME/conf/hive-site.xml

Spark若能直接访问MySQL中已有的元数据信息 $SPARK_HOME/conf/hive-site.xml

前置条件

在使用 Spark 整合 Hive 之前，需要安装配置以下软件：

• Hadoop：用于数据存储和分布式计算。
• Hive：用于数据存储和查询。
• Spark：用于分布式计算。

整合 Hive

在 Spark 中使用 Hive，需要将 Hive 的依赖库添加到 Spark 的类路径中。在 Java 代码中，可以使用 SparkConf 对象来设置 Spark 应用程序的配置。下面是一个示例代码：

import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

public class SparkHiveIntegration {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("SparkHiveIntegration")
                .setMaster("local[*]")
                .set("spark.sql.warehouse.dir", "/user/hive/warehouse");
        SparkSession spark = SparkSession.builder()
                .config(conf)
                .enableHiveSupport()
                .getOrCreate();
        spark.sql("SELECT * FROM mytable").show();
        spark.stop();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在上面的代码中，首先创建了一个 SparkConf 对象，设置了应用程序的名称、运行模式以及 Hive 的元数据存储路径。然后，创建了一个 SparkSession 对象，启用了 Hive 支持。最后，使用 Spark SQL 查询语句查询了一个名为 mytable 的 Hive 表，并将结果打印出来。最后，停止了 SparkSession 对象。

需要注意的是，Spark SQL 语法与 Hive SQL 语法略有不同，可以参考 Spark SQL 官方文档。

2 ThiriftServer使用

javaedge@JavaEdgedeMac-mini sbin % pwd
/Users/javaedge/Downloads/soft/spark-2.4.3-bin-2.6.0-cdh5.15.1/sbin

javaedge@JavaEdgedeMac-mini sbin % ./start-thriftserver.sh --master local --jars /Users/javaedge/.m2/repository/mysql/mysql-connector-java/8.0.15/mysql-connector-java-8.0.15.jar

starting org.apache.spark.sql.hive.thriftserver.HiveThriftServer2, logging to /Users/javaedge/Downloads/soft/spark-2.4.3-bin-2.6.0-cdh5.15.1/logs/spark-javaedge-org.apache.spark.sql.hive.thriftserver.HiveThriftServer2-1-JavaEdgedeMac-mini.local.out

1
2
3
4
5
6
7

beeline

内置了一个客户端工具：

javaedge@JavaEdgedeMac-mini bin % ./beeline -u jdbc:hive2://localhost:10000
Connecting to jdbc:hive2://localhost:10000
log4j:WARN No appenders could be found for logger (org.apache.hive.jdbc.Utils).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Connected to: Spark SQL (version 2.4.3)
Driver: Hive JDBC (version 1.2.1.spark2)
Transaction isolation: TRANSACTION_REPEATABLE_READ
Beeline version 1.2.1.spark2 by Apache Hive
0: jdbc:hive2://localhost:10000>
1
2
3
4
5
6
7
8
9
10

当你执行一条命令后：

就能在 Web UI 看到该命令记录：

3 通过代码访问数据

总是手敲命令行肯定太慢了，我们更多是代码访问：

package com.javaedge.bigdata.chapter06

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object JDBCClientApp {

  def main(args: Array[String]): Unit = {
    Class.forName("org.apache.hive.jdbc.HiveDriver")

    val conn: Connection = DriverManager.getConnection("jdbc:hive2://localhost:10000")
    val pstmt: PreparedStatement = conn.prepareStatement("show tables")

    val rs: ResultSet = pstmt.executeQuery()

    while (rs.next()) {
      println(rs.getObject(1) + " : " + rs.getObject(2))
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

最后打成 jar 包，扔到服务器定时运行即可执行作业啦。

ThiriftServer V.S Spark Application 例行作业

Thrift Server 独立的服务器应用程序，它允许多个客户端通过网络协议访问其上运行的 Thrift 服务。Thrift 服务通常是由一组 Thrift 定义文件定义的，这些文件描述了可以从客户端发送到服务器的请求和响应消息的数据结构和协议。Thrift Server 可以使用各种编程语言进行开发，包括 Java、C++、Python 等，并支持多种传输和序列化格式，例如 TSocket、TFramedTransport、TBinaryProtocol 等。使用 Thrift Server，您可以轻松地创建高性能、可伸缩和跨平台的分布式应用程序。

Spark Application，基于 Apache Spark 的应用程序，它使用 Spark 编写的 API 和库来处理大规模数据集。Spark Application 可以部署在本地计算机或云环境中，并且支持各种数据源和格式，如 Hadoop 分布式文件系统（HDFS）、Apache Cassandra、Apache Kafka 等。Spark Application 可以并行处理数据集，以加快数据处理速度，并提供了广泛的机器学习算法和图形处理功能。使用 Spark Application，您可以轻松地处理海量数据，提取有价值的信息和洞察，并帮助您做出更明智的业务决策。

因此，Thrift Server 和 Spark Application 适用不同的场景和应用程序：

• 需要创建一个分布式服务并为多个客户端提供接口，使用 Thrift Server
• 需要处理大规模数据集并使用分布式计算和机器学习算法来分析数据，使用 Spark Application

4 Spark 代码访问 Hive 数据

5 Spark SQL 函数实战

parallelize

SparkContext 一个方法，将一个本地数据集转为RDD。parallelize` 方法接受一个集合作为输入参数，并根据指定的并行度创建一个新的 RDD。

语法：

// data表示要转换为 RDD 的本地集合
// numSlices表示 RDD 的分区数，通常等于集群中可用的 CPU 核心数量。 
val rdd = 
sc.parallelize(data, numSlices)
1
2
3
4

将一个包含整数值的本地数组转换为RDD：

import org.apache.spark.{SparkConf, SparkContext}

// 创建 SparkConf 对象
val conf = new SparkConf().setAppName("ParallelizeExample").setMaster("local[*]")

// 创建 SparkContext 对象
val sc = new SparkContext(conf)

// 定义本地序列
val data = Seq(1, 2, 3, 4, 5)

// 使用 parallelize 方法创建 RDD
val rdd = sc.parallelize(data)

// 执行转换操作
val result = rdd.map(_ * 2)

// 显示输出结果
result.foreach(println)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

创建了一个包含整数值的本地序列 data，然后使用 parallelize 方法将其转换为一个 RDD。接下来，我们对 RDD 进行转换操作，并打印输出结果。

使用 parallelize 方法时，请确保正确配置 Spark 应用程序，并设置正确 CPU 核心数量和内存大小。否则，可能会导致应用程序性能下降或崩溃。

5.1 内置函数

都在这：

统计 PV、UV 实例

package com.javaedge.bigdata.chapter06

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 内置函数
 */
object BuiltFunctionApp {

  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession.builder()
      .master("local").appName("HiveSourceApp")
      .getOrCreate()

    // day  userid
    val userAccessLog = Array(
      "2016-10-01,1122",
      "2016-10-01,1122",
      "2016-10-01,1123",
      "2016-10-01,1124",
      "2016-10-01,1124",
      "2016-10-02,1122",
      "2016-10-02,1121",
      "2016-10-02,1123",
      "2016-10-02,1123"
    )

    import spark.implicits._

    // Array ==> RDD
    val userAccessRDD: RDD[String] = spark.sparkContext.parallelize(userAccessLog)

    val userAccessDF: DataFrame = userAccessRDD.map(x => {
      val splits: Array[String] = x.split(",")
      Log(splits(0), splits(1).toInt)
    }).toDF

    userAccessDF.show()

    import org.apache.spark.sql.functions._

    // select day, count(user_id) from xxx group by day;
    userAccessDF.groupBy("day").agg(count("userId").as("pv")).show()

    userAccessDF.groupBy("day").agg(countDistinct("userId").as("uv")).show()
    spark.stop()
  }

  private case class Log(day: String, userId: Int)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

5.2 自定义函数

package com.javaedge.bigdata.chapter06

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}


/**
 * 统计每个人爱好的个数
 * pk：3
 * jepson： 2
 *
 *
 * 1）定义函数
 * 2）注册函数
 * 3）使用函数
 */
object UDFFunctionApp {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local").appName("HiveSourceApp")
      .getOrCreate()


    import spark.implicits._

    val infoRDD: RDD[String] = spark.sparkContext.textFile(
      "/Users/javaedge/Downloads/sparksql-train/data/hobbies.txt")
    val infoDF: DataFrame = infoRDD.map(_.split("###")).map(x => {
      Hobbies(x(0), x(1))
    }).toDF

    infoDF.show(false)

    // TODO... 定义函数 和 注册函数
    spark.udf.register("hobby_num", (s: String) => s.split(",").size)

    infoDF.createOrReplaceTempView("hobbies")

    //TODO... 函数的使用
    spark.sql("select name, hobbies, hobby_num(hobbies) as hobby_count from hobbies").show(false)

    // select name, hobby_num(hobbies) from xxx

    spark.stop()
  }

  private case class Hobbies(name: String, hobbies: String)
}

output：
+------+----------------------+
|name  |hobbies               |
+------+----------------------+
|pk    |jogging,Coding,cooking|
|jepson|travel,dance          |
+------+----------------------+

+------+----------------------+-----------+
|name  |hobbies               |hobby_count|
+------+----------------------+-----------+
|pk    |jogging,Coding,cooking|3          |
|jepson|travel,dance          |2          |
+------+----------------------+-----------+
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

6 总结

通过上述示例代码，可以看到如何在 Java 中使用 Spark 整合 Hive。通过使用 Hive 的数据存储和查询功能，可以在 Spark 中高效地处理和分析数据。当然，还有许多其他功能和配置可以使用，例如设置 Spark 应用程序的资源分配、数据分区、数据格式转换等等。