Spark Learning

Tao Zou

2025-06-10

Spark安装

环境变量：

#SPARK_HOME
export SPARK_HOME=/opt/Spark355/spark-3.5.5-bin-hadoop3
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

验证Spark已经配置到环境中：

spark-shell --version

Spark识别Hadoop集群

Spark的Kerberos安全模式

我的Hadoop集群采用Kerberos安全模式。

参照创建Kerberos的Kafka服务主体和用户主体创建Kerberos的Spark服务主体和用户主体。

useradd -g hadoop -m spark_user
# 在Kerberos服务端执行下面这一行命令创建Spark用户主体并设置好密码
kadmin.local -q "add_principal spark_user"
# 另外再为kafka_user创建一个keytab（暂且创建在hadoop102上）
kadmin.local -q "ktadd -kt /etc/security/keytabs/sparkuserService.keytab spark_user@EXAMPLE.COM"
chown spark_user:hadoop /etc/security/keytabs/sparkuserService.keytab
cp /etc/security/keytabs/sparkuserService.keytab $SPARK_HOME/conf/kerberos

# 在每个节点上执行下面一行命令，并且设置不同节点的kafka_user用户两两之间ssh免密
useradd -g hadoop -m spark_user

（后面再研究这里如何使用kerberos安全模式）

Spark on Yarn

Spark的Driver运行在Yarn提供的容器内或提交任务的客户端进程中；Spark的Executor的容器运行在Yarn提供的容器内。

建立Spark与Hadoop集群的连接：

echo -e "export HADOOP_CONF_DIR=/opt/Hadoop336/hadoop-3.3.6/etc/hadoop\nexport YARN_CONF_DIR=/opt/Hadoop336/hadoop-3.3.6/etc/hadoop" >> $SPARK_HOME/conf/spark-env.sh

之后在Spark服务端节点的mapred_user用户（拿到了jhs票据）下执行$SPARK_HOME/bin/pyspark --master yarn启动pyspark on yarn的客户端。

编译器启动Spark

使用IntelliJ Ultimate版直连服务器。

Spark-Java

准备工作：

在启动Spark-Java之前一定要提前在core-site.xml文件中添加Keberos的映射规则：RULE:[1:$1@$0](mapred_user@EXAMPLE\.COM)s/.*/mapred_user/，不然程序会将用户识别为mapred_user@EXAMPLE.COM而不是mapred_user。
将core-site.xml，hdfs-site.xml，yarn-site.xml, mapred-site.xml全部复制到maven项目的resources目录中。
添加运行环境变量：HADOOP_CONF_DIR=/opt/Hadoop336/hadoop-3.3.6/etc/hadoop;YARN_CONF_DIR=/opt/Hadoop336/hadoop-3.3.6/etc/hadoop

pom.xml

在pom.xml中加入下述依赖：

    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>2.0.12</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.5.5</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-yarn_2.12</artifactId>
            <version>3.5.5</version>
        </dependency>

        <!-- 这个依赖的作用是使Spark UI能正常启动，否则会报错找不到类org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-yarn-server-web-proxy</artifactId>
            <version>3.3.6</version>
        </dependency>
    </dependencies>

Spark-Python

SparkSQL (Spark on Hive)

添加依赖

        <!-- 下面这个依赖用于Spark SQL -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.5.5</version>
        </dependency>

SparkSQL connecting to Hive

使用Spark来解析SQL来操作Hive数据库。

环境准备

确保启动Hive的metastore服务：

nohup hive --service metastore > /tmp/metastore.log 2>&1 &

将$HIVE_HOME/conf/hive-site.xml拷贝到maven项目的resources目录下。
添加下述依赖

        <!-- 用于Spark SQL连接Hive数据源 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.5.5</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.31</version>
        </dependency>

代码实现

final SparkSession sparkSession = SparkSession
        .builder()
        .config(conf)
        .enableHiveSupport()
        .getOrCreate();
sparkSession.sql("SHOW TABLES;").show();  // 打印出Hive数据库中的数据表

Hive

Hive安装

下载Hive-4.0.1：(下面这个网站国内下载速度非常慢)

wget https://dlcdn.apache.org/hive/hive-4.0.1/apache-hive-4.0.1-bin.tar.gz

环境变量：

#HIVE_HOME
export HIVE_HOME=/opt/Hive401/apache-hive-4.0.1-bin
export PATH=$PATH:$HIVE_HOME/bin

下述命令的作用是解决jar包日志冲突。

cd $HIVE_HOME/lib
mv log4j-slf4j-impl-2.18.0.jar log4j-slf4j-impl-2.18.0.jar.bak

将MySQL的JDBC驱动拷贝到 $HIVE_HOME/lib$ 下。

wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-j-8.0.31.tar.gz

解压后找到文件mysql-connector-j-8.0.31.jar

cp mysql-connector-j-8.0.31.jar $HIVE_HOME/lib

配置hive-site.xml

vim $HIVE_HOME/conf/hive-site.xml

配置Metastore到HIVE：

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
        <!-- 配置Hive要连接的数据库地址。这是远程连接，如果远程连接root用户好像会报错，所以我连接的是zt@%用户。 -->
        <property>
                <name>javax.jdo.option.ConnectionURL</name>
                <value>jdbc:mysql://hadoop102:3306/metastore?useSSL=false&amp;useUnicode=true&amp;characterEncoding=UTF-8&amp;allowPublicKeyRetrieval=true</value>
        </property>

        <property>
                <name>javax.jdo.option.ConnectionDriverName</name>
                <value>com.mysql.cj.jdbc.Driver</value>
        </property>
        <!-- 配置Hive连接MySQL的用户名 -->
        <property>
                <name>javax.jdo.option.ConnectionUserName</name>
                <value>zt</value>
        </property>

        <!-- 配置Hive连接MySQL的用户密码 -->
        <property>
                <name>javax.jdo.option.ConnectionPassword</name>
                <value>123456</value>
        </property>

        <!-- HDFS上Hive数据的保存路径 -->
        <property>
                <name>hive.metastore.warehouse.dir</name>
                <value>/user/hive/warehouse</value>
        </property>

        <property>
                <name>hive.metastore.schema.verification</name>
                <value>false</value>
        </property>

        <property>
                <name>hive.server2.thrift.port</name>
                <value>10000</value>
        </property>

        <property>
                <name>hive.server2.thrift.bind.host</name>
                <value>hadoop102</value>
        </property>

        <property>
                <name>hive.cli.print.header</name>
                <value>true</value>
        </property>

        <property>
                <name>hive.cli.print.current.db</name>
                <value>true</value>
        </property>
</configuration>

配置Hive日志路径

mkdir $HIVE_HOME/logs
mv $HIVE_HOME/conf/hive-log4j2.properties.template $HIVE_HOME/conf/hive-log4j2.properties
vim $HIVE_HOME/conf/hive-log4j2.properties

修改下述内容：

property.hive.log.dir = /opt/Hive401/apache-hive-4.0.1-bin/logs

初始化Hive元数据库

使用root用户创建数据库metastore，然后将该数据库的权限给用户zt@%

初始化Hive元数据库（Hive3的做法）：

cd $HIVE_HOME
schematool -initSchema -dbType mysql -verbose

初始化Hive元数据库（Hive4的做法）：

cd $HIVE_HOME
mysql -u zt -p mypassword metastore < $HIVE_HOME/scripts/metastore/upgrade/mysql/hive-schema-4.0.0.mysql.sql
beeline -u jdbc:mysql://hadoop102:3306/metastore -n zt -p mypassword

修改原数据库metastore字符集。

-- 使用zt@%登录MySQL
USE metastore;
ALTER TABLE COLUMNS_V2 modify column COMMENT VARCHAR(256) character SET utf8;
ALTER TABLE TABLE_PARAMS modify column PARAM_VALUE mediumtext character SET utf8;

启动Hive4.x

Hive4.x跟以前版本的启动有很大不同。

由于我对Hive的笔记梳理得很混乱，这里暂时记录目前行之有效的Hive4.x启动方式。

启动Kerberos安全模式下的 hadoop集群。
在root用户下运行hive --service hiveserver2 &来后台启动hiveserver2服务。
在root用户下运行hive直接打开beeline命令行，然后输入命令!connect jdbc:hive2://hadoop102:10000即可直接与hiveserver2建立连接，然后就可以直接编写和运行SQL代码了。

可能由于之前已经配置好了hive的kerberos相关配置，上述过程会自动使用我的hive_user完成kerberos安全认证，并直接调用yarn执行mapreduce任务。

Hive on Spark

Hive默认使用的计算引擎是MapReduce。现在将其替换为Spark（Hive on Spark）。

Hive-3.1.3与Spark-3.3.0是默认不兼容的，为了使两者兼容，下面开始重新编译Hive-3.1.3。

（具体编译方法以后再尝试）

Hive处理Json

将HDFS上的Json读取到Hive中的自建表myjson_table。

CREATE EXTERNAL TABLE myjson_table (
json_txt STRING
) LOCATION "/financeApproval/tmp";

在Hive中解析myjson_table

SELECT 
get_json_object(json_txt, '$.database') `database_col`,
get_json_object(json_txt, '$.table') `table_col`,
get_json_object(json_txt, '$.type') `type_col`,
get_json_object(json_txt, '$.ts') `ts_col`,
get_json_object(json_txt, '$.data.id') `id_col`,
get_json_object(json_txt, '$.data.name') `name_col`
FROM myjson_table;

安装和配置Spark-3.3.2

安装：

wget https://archive.apache.org/dist/spark/spark-3.3.2/spark-3.3.2-bin-without-hadoop.tgz
# 华为云镜像
wget https://mirrors.huaweicloud.com/apache/spark/spark-3.3.2/spark-3.3.2-bin-without-hadoop.tgz

配置：

mv /opt/Spark332/spark-3.3.2-bin-without-hadoop/conf/spark-env.sh.template /opt/Spark332/spark-3.3.2-bin-without-hadoop/conf/spark-env.sh
vim /opt/Spark332/spark-3.3.2-bin-without-hadoop/conf/spark-env.sh

追加下述内容：

export SPARK_DIST_CLASSPATH=$(hadoop classpath)

添加SPARK_HOME环境变量

#SPARK_HOME
export SPARK_HOME=/opt/Spark332/spark-3.3.2-bin-without-hadoop
export PATH=$PATH:$SPARK_HOME/bin

在Hive中配置Spark

vim /opt/Hive313/apache-hive-3.1.3-bin/conf/spark-defaults.conf

spark.master                    yarn
spark.eventLog.enabled          true
spark.eventLog.dir              hdfs://hadoop102:8020/spark-history
spark.executor.memory           1g
spark.driver.memory             1g

然后在HDFS上创建文件夹：hadoop fs -mkdir /spark-history

Hive配置Spark依赖和Spark引擎

先向HDFS上环Spark的所有依赖：

hadoop fs -mkdir /spark-jars
hadoop fs -put $SPARK_HOME/jars/* /spark-jars

为Hive配置这些依赖：

vim /opt/Hive313/apache-hive-3.1.3-bin/conf/hive-site.xml

    <!-- Spark依赖的位置 -->
    <property>
        <name>spark.yarn.jars</name>
        <value>hdfs://hadoop102:8020/spark-jars/*</value>
    </property>
    <!-- 指定引擎为Spark -->
    <property>
        <name>hive.execution.engine</name>
        <value>spark</value>
    </property>
    <!-- 设置默认任务超时的时间，单位毫秒-->
    <property>
        <name>hive.spark.client.connect.timeout</name>
        <value>5000</value>
    </property>

Kerberos安全模式

准备Principal和linux用户，并配置权限：

# 在每个节点上都要执行下面这一行创建hive_user的代码，并相互设置免密ssh登录
useradd -g hadoop -m hive_user 

chown -R hive_user:hadoop /opt/Hive313/apache-hive-3.1.3-bin
kadmin.local -q "add_principal hive_user"

kadmin.local -q "add_principal -randkey hive/hadoop102@EXAMPLE.COM"
kadmin.local -q "ktadd -kt /etc/security/keytabs/hiveService.keytab hive/hadoop102@EXAMPLE.COM"
chown hive_user:hadoop /etc/security/keytabs/hiveService.keytab
chmod 400 /etc/security/keytabs/hiveService.keytab

修改hive-env.sh文件

mv /opt/Hive313/apache-hive-3.1.3-bin/conf/hive-env.sh.template /opt/Hive313/apache-hive-3.1.3-bin/conf/hive-env.sh
vim /opt/Hive313/apache-hive-3.1.3-bin/conf/hive-env.sh

追加下述内容：

export JAVA_HOME=/opt/JDK8/jdk1.8.0_202
export HADOOP_HOME=/opt/Hadoop336/hadoop-3.3.6
export HIVE_HOME=/opt/Hive313/apache-hive-3.1.3-bin
export HIVE_CONF_DIR=/opt/Hive313/apache-hive-3.1.3-bin/conf

修改hive-site.xml文件

vim /opt/Hive313/apache-hive-3.1.3-bin/conf/hive-site.xml

添加下述Kerberos安全相关配置：

    <property>
        <name>hive.server2.authentication</name>
        <value>KERBEROS</value>
    </property>
    <property>
        <name>hive.server2.authentication.kerberos.principal</name>
        <value>hive/_HOST@EXAMPLE.COM</value>
    </property>
    <property>
        <name>hive.server2.authentication.kerberos.keytab</name>
        <value>/etc/security/keytabs/hiveService.keytab</value>
    </property>

    <property>
        <name>hive.metastore.sasl.enabled</name>
        <value>true</value>
    </property>
    <property>
        <name>hive.metastore.kerberos.principal</name>
        <value>hive/_HOST@EXAMPLE.COM</value>
    </property>
    <property>
        <name>hive.metastore.kerberos.keytab.file</name>
        <value>/etc/security/keytabs/hiveService.keytab</value>
    </property>

上述内容配置好后，执行hive后，系统当前的票据会自动变成hive/hadoop102@EXAMPLE.COM。

在$HADOOP_HOME/etc/hadoop/core-site.xml中新增hive_user的映射规则，以及允许hive在任何主机上模拟属于hadoop组的用户：

    <property>
        <name>hadoop.security.auth_to_local</name>
        <value>
            RULE:[2:$1/$2@$0]([dn]n\/.*@EXAMPLE\.COM)s/.*/hdfs_user/
            RULE:[2:$1/$2@$0](snn\/.*@EXAMPLE\.COM)s/.*/hdfs_user/
            RULE:[2:$1/$2@$0]([rn]m\/.*@EXAMPLE\.COM)s/.*/yarn_user/
            RULE:[2:$1/$2@$0](jhs\/.*@EXAMPLE\.COM)s/.*/mapred_user/
            RULE:[2:$1/$2@$0](hive\/.*@EXAMPLE\.COM)s/.*/hive_user/
            RULE:[1:$1@$0](zt@EXAMPLE\.COM)s/.*/zt/
        </value>
    </property>

<!-- 按照视频里的教程，这里还加了一堆hive, hdfs, HTTP的* -->

分发到所有节点。

Hive DDL

数据库

创建数据库

CREATE DATABASE [IF NOT EXISTS] mydb
[COMMENT 'This is my first DB in Hive']
[LOCATION hdfs_path]
[WITH DBPROPERTIES (property_name=property_value, ...)];

LOCATION若不指定，则默认路径为：${hive.metastore.warehouse.dir}/database_name.db。
WITH DBPROPERTIES是给数据库附加属性的，例如追加该数据库的负责人或数据库创建时间，例如：WITH DBPROPERTIES(‘create_date’=‘2022-11-18’)。

查询数据库

SHOW DATABASES [LIKE 'this is wildcard'];

描述某个数据库

DESCRIBE DATABASE [EXTENDED] db_name;

EXTENDED表示展示更为详细的信息。

修改数据库

ALTER DATABASE database_name SET DBPROPERTIES (property_name=property_value, ...);

ALTER DATABASE database_name SET LOCATION hdfs_path;

ALTER DATABASE database_name SET OWNER USER user_name;

更改数据库的路径不会影响已存在的数据库中的表的路径，只会影响新表的默认路径。

删除数据库

DROP DATABASE [IF EXISTS] database_name [RESTRICT|CASCADE];

RESTRICT为默认的严格模式，数据库中存在表则数据库删除失败；CASCADE为级联模式，数据库中的表会被一并删除。

数据表

创建表

CREATE TABLE [IF NOT EXISTS] [db_name.]mytable AS SELECT ...

CREATE TABLE [IF NOT EXISTS] [db_name.]mytable LIKE another_table;