Kerberos 认证

本集群启用了Kerberos 认证，使用 Spark 客户端前需先完成认证。管理员可以使用集群成员服务器上的 spark keytab 进行 Kerberos 认证，从而获得 Spark 集群操作权限，用户需要先进行认证才能对集群进行操作，否则会报认证失败异常。

在命令行中输入如下命令即可完成Kerberos 认证：

klist -kt /etc/security/keytabs/spark.keytab | awk '{print $4}' | grep -e 'spark.*@.*' | head -n 1
kinit -kt /etc/security/keytabs/spark.keytab <principal>

其中klist -kt 输出的第 4 列为 principal（主体名），通过 grep 过滤出 spark 用户的 principal 后用于 kinit 认证。

认证完成后可通过klist 命令查看当前票据信息，票据过期需重新认证。

spark-submit 提交作业

spark-submit 是 Spark 作业的统一提交入口，可提交 Java/Scala（jar 包）、Python 脚本等应用，基本用法如下：

spark-submit \
  --master yarn \
  --deploy-mode client \
  --class <main-class> \
  --name <application-name> \
  --queue <yarn-queue> \
  --num-executors <num> \
  --executor-memory <size> \
  --executor-cores <num> \
  --driver-memory <size> \
  --conf <key>=<value> \
  <application-jar> [application-arguments]

主要参数说明如下表所示：

运行示例：SparkPi

使用Spark 自带示例程序验证集群可用性，在命令行中输入如下命令：

spark-submit --master yarn --deploy-mode client \
  --class org.apache.spark.examples.SparkPi \
  $SPARK_HOME/examples/jars/spark-examples_2.12-3.4.1.jar 100

执行成功后终端输出类似如下内容：

Pi is roughly 3.1418599285819928

可通过yarn application -list 查看 application 状态为 FINISHED：

yarn application -list -appStates FINISHED

spark-shell / pyspark 交互式环境

spark-shell 提供基于 Scala 的交互式编程环境，pyspark 提供基于 Python 的交互式环境，两者均在启动时自动创建 SparkSession（变量名 spark），适合交互式探索与调试。启动命令如下：

spark-shell --master yarn --deploy-mode client
pyspark --master yarn --deploy-mode client

spark-shell 中执行 WordCount 示例：

val text = spark.read.textFile("/tmp/input.txt")
val counts = text.flatMap(line => line.split(" ")).groupBy("value").count()
counts.show()

pyspark 中执行简单 SQL 示例：

df = spark.createDataFrame([("a", 1), ("b", 2)], ["k", "v"])
df.createOrReplaceTempView("t")
spark.sql("select k, sum(v) from t group by k").show()

spark-sql CLI

Spark SQL CLI 是一个交互式命令行客户端工具，可以连接 Hive 元存储服务，并执行输入的 SQL。常用方式为通过 -e 执行单条语句或 -f 执行脚本文件：

spark-sql -e "select count(*) from db_name.table_name"
spark-sql -f /path/to/script.sql   # script.sql 为用户自建的SQL脚本文件

spark-sql 的主要启动参数如下表所示（spark-sql --help 可查看完整列表，包含【options】与【cli option】两类，前者为 Spark 多客户端共用参数，后者为 Spark SQL CLI 独有参数）：

不带-e 或 -f 参数时，spark-sql 进入交互式 shell 模式，以分号（;）分隔命令。输入 quit; 或 exit; 退出客户端。交互模式下的常用命令如下表所示：

Spark SQL 语法支持

Spark SQL 语法分为 DDL、DML、数据检索（Data Retrieval）与辅助（Auxiliary）四类，主要语法清单如下表所示：

各语法的使用细节可参考Spark 官方文档的 SQL 语法章节。

Spark SQL 示例：企业雇员分析

以企业雇员分析为背景构建三张表：雇员基本信息表（employees_info）、雇员通信表（employees_contact）、公司部门表（department），演示建库建表、导入数据与分析查询的完整流程。

1、创建数据库及表：

CREATE DATABASE IF NOT EXISTS company;
USE company;

CREATE TABLE IF NOT EXISTS employees_info
(
  id INT COMMENT '雇员ID',
  name STRING COMMENT '雇员名称',
  gender STRING COMMENT '性别',
  dept_id INT COMMENT '部门ID',
  grade STRING COMMENT '雇员工资等级',
  salary DOUBLE COMMENT '雇员薪水',
  deduction_rate DOUBLE COMMENT '雇员税率',
  entry_date STRING COMMENT '入职日期',
  other_info STRING COMMENT '额外信息'
)
USING PARQUET
COMMENT '员工信息基础表';

CREATE TABLE IF NOT EXISTS employees_contact
(
  id INT COMMENT '雇员ID',
  phone STRING COMMENT '雇员电话号',
  email STRING COMMENT '雇员email'
)
USING PARQUET
COMMENT '雇员通信表';

CREATE TABLE IF NOT EXISTS department
(
  dept_id INT COMMENT '部门ID',
  dept_name STRING COMMENT '部门名称'
);

2、导入数据（示例数据节选）：

INSERT INTO employees_contact VALUES
  (53275, '145939987', 'oqjunox@yahoo.com'),
  (21430, '183727019', 'tmmprglpsy@gmail.com'),
  (79254, '143213819', 'nxtay@outlook.com');

INSERT INTO employees_info VALUES
  (7821, 'Meryl Wilson', 'm', 1, 'M1', 120129, 0.23, '2007-4-8',
   '{"country": "china", "city": "beijing", "single": false}'),
  (9130, 'Jacob Tracie', 'f', 2, 'M2', 15305, 0.53, '2002-2-11',
   '{"country": "china", "city": "beijing", "single": false}');

INSERT INTO department VALUES (1, '测试'), (2, '开发'), (3, '产品');

3、分析查询：

-- 查看各个部门雇员薪水的平均值
SELECT dept.dept_name, ROUND(AVG(emp.salary), 2)
FROM employees_info emp
JOIN department dept ON emp.dept_id = dept.dept_id
GROUP BY dept.dept_name;

-- 查看雇员中各个号段的数量
SELECT SUBSTRING(phone, 1, 3) AS phone_field, COUNT(phone) AS cnt
FROM employees_contact
GROUP BY SUBSTRING(phone, 1, 3);

-- 查看雇员额外信息中包含哪些内容
SELECT name, other_info AS info FROM employees_info;

Scala SDK 开发

Spark 支持 Java/Scala/Python/R 语言进行应用开发。由于 Spark 本身主要由 Scala 语言开发，且 Scala 语言简洁易懂，推荐使用 Scala 语言开发。向工程 pom 文件添加依赖（版本与本集群一致，scala 为 2.12）：

<properties>
  <spark.version>3.4.1</spark.version>
  <scala.version>2.12</scala.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_${scala.version}</artifactId>
    <version>${spark.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_${scala.version}</artifactId>
    <version>${spark.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_${scala.version}</artifactId>
    <version>${spark.version}</version>
  </dependency>
</dependencies>

{"name":"Alice", "age":10}
{"name":"Jessica", "age":23}
{"name":"Albert", "age":16}
{"name":"Anna", "age":55}

新建object 文件 SparkSQLExample：

package spark_sql_demo

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

object SparkSQLExample {
  case class Person(name: String, age: Long)

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
    val spark = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .config(sparkConf)
      .getOrCreate()
    runExample(spark)
    spark.stop()
  }

  private def runExample(spark: SparkSession): Unit = {
    val df = spark.read.json("src/main/resources/people.json")
    df.show()

    import spark.implicits._
    df.filter($"age" >= 18).show()   // 筛选成年人

    df.createOrReplaceTempView("people")
    spark.sql("SELECT count(*) FROM people").show()  // 统计总人数
  }
}

集群运行完整步骤（从零开始）：

1、将上文的 4 行人员数据保存为本地文件 people.json；

2、上传到 HDFS（用户目录下，命令会自动创建 /user/<用户名>/people.json）：

hdfs dfs -put people.json /user/<用户名>/  # 或简化为: hdfs dfs -put people.json people.json

3、修改代码中的读取路径为 HDFS 路径：

val df = spark.read.json("hdfs:///user/<用户名>/people.json")

4、maven 打包获得 spark_sql_demo-1.0-SNAPSHOT.jar，通过 spark-submit 运行：

spark-submit --class spark_sql_demo.SparkSQLExample \
spark_sql_demo-1.0-SNAPSHOT.jar

配置说明

Spark 的服务端与客户端配置文件位于 $SPARK_HOME/conf 目录下，主要配置文件说明如下表所示：

常用配置参数及其默认值如下表所示（取自Spark 3.4.1 源码默认配置）：