Apache Gluten 快速入门与集成示例

Apache Gluten 作为 Spark 的插件运行，业务 SQL/DataFrame 代码无需任何改动即可把算子下推到原生引擎。本文从零带你跑通第一个查询，再给出在 Spark 应用、SparkConf 与 Spark Thrift Server 中集成 Gluten 的完整示例，所有输出均来自 Apache Gluten 1.7.0 + Spark 3.5.5 + JDK 17 的真实运行。

一、快速入门

1.1 环境准备

假设 JDK 17 与 Spark 3.5.5 已就绪，Gluten bundle 已放到 /opt/gluten/。设置基础变量：

export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64   # CentOS 8 为 /usr/lib/jvm/java-17-openjdk
export PATH=$JAVA_HOME/bin:$PATH
export SPARK_HOME=/opt/spark
GLUTEN_JAR=/opt/gluten/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.7.0.jar
JVM_OPTS='--add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED -Dio.netty.tryReflectionSetAccessible=true'

其中 JVM_OPTS 是 JDK 17 下的必需项。

1.2 第一个查询

准备 SQL 文件：

-- /root/test.sql
SELECT count(*) AS cnt, sum(id) AS s FROM range(1000000);

运行：

$SPARK_HOME/bin/spark-sql --master 'local[4]' --driver-memory 512m \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=1g \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
  --conf "spark.driver.extraJavaOptions=$JVM_OPTS" \
  --conf "spark.executor.extraJavaOptions=$JVM_OPTS" \
  --jars $GLUTEN_JAR -f /root/test.sql

真实输出：

1000000 499999500000

1.3 输出解读

结果 1000000 499999500000 表示 1..1,000,000 共 1,000,000 行，求和为 499,999,500,000，与数学期望一致；
这条查询中 count/sum 已由 Velox 原生算子执行，GlutenPlugin 负责把物理计划转换后交给原生引擎，再把列式结果转回 Spark 行格式；
整个过程中业务 SQL 未做任何修改，只是多了几个 --conf。

1.4 如何确认原生已生效

把 org.apache.gluten 日志级别设为 INFO，启动日志中应出现：

INFO GlutenDriverPlugin: Gluten components:
Component velox
INFO JniLibLoader: Successfully loaded library linux/amd64/libgluten.so
INFO JniLibLoader: Successfully loaded library linux/amd64/libvelox.so
INFO VeloxListenerApi: Gluten is running with Spark local mode. Skip running static initializer for executor.
INFO VeloxBackend: Gluten SQL Tab has been attached.

二、集成方式

2.1 命令行参数（spark-submit / spark-sql / spark-shell）

在原有提交命令上追加 --jars 与若干 --conf 即可，见 1.2。

2.2 代码内 SparkConf（Scala）

import org.apache.spark.sql.SparkSession

val conf = new org.apache.spark.SparkConf()
  .setAppName("GlutenDemo")
  .setMaster("local[4]")
  .set("spark.plugins", "org.apache.gluten.GlutenPlugin")
  .set("spark.memory.offHeap.enabled", "true")
  .set("spark.memory.offHeap.size", "1g")
  .set("spark.shuffle.manager", "org.apache.spark.shuffle.sort.ColumnarShuffleManager")
  .set("spark.driver.extraJavaOptions",
       "--add-opens=java.base/java.nio=ALL-UNNAMED " +
       "--add-opens=java.base/java.lang=ALL-UNNAMED " +
       "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED " +
       "-Dio.netty.tryReflectionSetAccessible=true")

val spark = SparkSession.builder().config(conf).getOrCreate()
println(spark.sql("SELECT count(*) AS c, sum(id) AS s FROM range(1000000)").collect().mkString)
spark.stop()

打包后用 --jars $GLUTEN_JAR 提交即可。

2.3 全局配置 spark-defaults.conf

在 $SPARK_HOME/conf/spark-defaults.conf 中追加，可让该 Spark 安装的所有作业默认启用：

spark.plugins                 org.apache.gluten.GlutenPlugin
spark.memory.offHeap.enabled  true
spark.memory.offHeap.size     2g
spark.shuffle.manager         org.apache.spark.shuffle.sort.ColumnarShuffleManager
spark.driver.extraClassPath   /opt/gluten/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.7.0.jar
spark.executor.extraClassPath /opt/gluten/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.7.0.jar

三、集成示例一：聚合分析

3.1 样例代码

// AggDemo.scala
import org.apache.spark.sql.SparkSession

object AggDemo {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("GlutenAggDemo")
      .getOrCreate()
    val df = spark.range(20000000).toDF("id")
    df.createOrReplaceTempView("t")
    val r = spark.sql(
      "SELECT count(*) AS cnt, sum(id) AS s FROM t WHERE id % 7 = 0").collect()
    println("RESULT=" + r.mkString)
    spark.stop()
  }
}

3.2 运行命令

$SPARK_HOME/bin/spark-submit --master 'local[4]' --driver-memory 512m \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=2g \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
  --conf "spark.driver.extraJavaOptions=$JVM_OPTS" \
  --jars $GLUTEN_JAR \
  --class AggDemo /opt/app/gluten-demo.jar

3.3 输出

RESULT=[2857143,28571421428571]

四、集成示例二：Spark Thrift Server（JDBC）

Gluten 同样适用于常驻的 Thrift Server，供 BI/SQL 客户端通过 JDBC 接入。

4.1 启动命令

$SPARK_HOME/sbin/start-thriftserver.sh --master 'local[4]' --driver-memory 512m \
  --hiveconf hive.server2.thrift.port=10000 \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=1g \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
  --conf "spark.driver.extraJavaOptions=$JVM_OPTS" \
  --jars $GLUTEN_JAR

启动日志中可见 Gluten 已挂载：

INFO JniLibLoader: Successfully loaded library linux/amd64/libgluten.so
INFO JniLibLoader: Successfully loaded library linux/amd64/libvelox.so
INFO VeloxBackend: Gluten SQL Tab has been attached.
INFO SparkShimLoader: Using Shim provider: List(org.apache.gluten.sql.shims.spark35.SparkShimProvider@...)

4.2 通过 beeline 查询

$SPARK_HOME/bin/beeline -u 'jdbc:hive2://localhost:10000' -n root \
  -e 'SELECT count(*) AS cnt, sum(id) AS s FROM range(100000);'

真实输出：

Connected to: Spark SQL (version 3.5.5)
Driver: Hive JDBC (version 2.3.9)
+---------+-------------+
|   cnt   |      s      |
+---------+-------------+
| 100000  | 4999950000  |
+---------+-------------+
1 row selected (3.487 seconds)

4.3 验证 Thrift 侧下推

在 beeline 中临时关闭 AQE 后查看计划，可见同样的原生算子：

SET spark.sql.adaptive.enabled=false;
EXPLAIN EXTENDED SELECT count(*) AS cnt, sum(id) AS s FROM range(100000);

VeloxColumnarToRow
+- ^(2) HashAggregateTransformer(keys=[], functions=[count(1), sum(id#42L)], isStreamingAgg=false, ...)
   +- ColumnarExchange SinglePartition, ENSURE_REQUIREMENTS, ... , [shuffle_writer_type=hash], ...
      +- VeloxResizeBatches 1024, 2147483647, 10485760
         +- ColumnarRange 0, 100000, 1, 4, 100000, [id#42L]

停止服务：

$SPARK_HOME/sbin/stop-thriftserver.sh

五、启用前后对照

5.1 执行计划对照

同一查询 SELECT count(*), sum(id) FROM range(1000000)，关闭 AQE 观察：

环节	未启用 Gluten	启用 Gluten
扫描	`Range (0, 1000000, step=1, splits=4)`	`ColumnarRange` / `ArrowColumnarToVeloxColumnar`
聚合	`HashAggregate`（partial/final）	`FlushableHashAggregateTransformer` / `HashAggregateTransformer`
交换	`Exchange SinglePartition`	`ColumnarExchange` + `VeloxResizeBatches`
输出	`Row` 格式	`VeloxColumnarToRow`

启用 Gluten 后，计划中出现 VeloxColumnarToRow、HashAggregateTransformer、ColumnarExchange、ColumnarRange 等节点，表示算子已下推原生引擎。

5.2 性能对照

环境：Spark 3.5.5 / JDK 17 / local[4] / driver 512m / offHeap 2g，取多轮稳定值。

查询	Gluten 开启	Gluten 关闭
字符串聚合（2000 万行）	2105 ms	793 ms
自连接聚合（800 万 × 800 万）	1479 ms	4348 ms

结论：连接/Shuffle 密集型（自连接）负载下 Gluten 约快 2.9×；而轻量字符串聚合的原生 Spark codegen 反而更快，说明 Gluten 的收益与工作负载强相关，简单查询会因 JNI 与列式搬运产生额外开销。

六、常见问题

Q1：启动报 UnsupportedClassVersionError ... class file version 61.0 怎么办？ Gluten 1.7.0 需要 JDK 17，请将 JAVA_HOME 指向 JDK 17 后重启。

Q2：报 sun.misc.Unsafe or java.nio.DirectByteBuffer.<init>(long, int) not available 怎么办？ 这是 JDK 17 未开放内部 API 所致，需为 driver/executor 追加 --add-opens=java.base/java.nio=ALL-UNNAMED、--add-opens=java.base/java.lang=ALL-UNNAMED、--add-opens=java.base/sun.nio.ch=ALL-UNNAMED 及 -Dio.netty.tryReflectionSetAccessible=true。

Q3：EXPLAIN 看不到 Velox 算子？ AQE 开启时初始物理计划仍是朴素 Spark 计划，请临时设置 spark.sql.adaptive.enabled=false 后再 EXPLAIN EXTENDED。

Q4：出现 Excluding runtime-incompatible component: velox-iceberg/delta/... 是否异常？ 这是 Gluten 发现 classpath 中缺少对应数据源依赖时的正常告警，不影响核心算子下推。

Q5：查询比原生 Spark 还慢？ Gluten 的收益依赖工作负载。对于极轻量的聚合或表达式，JNI 调用与行列转换的开销可能超过收益；可通过 spark.gluten.sql.columnar.query.fallback.threshold 等回退参数控制边界。