Spark大数据开发与应用案例(视频教学版)（十七）--第十一章

辉哥大数据

982人浏览 · 2025-10-12 18:48:37

辉哥大数据 · 2025-10-12 18:48:37 发布

作者信息

作者：余辉微信公众号：辉哥大数据
购买地址： 京东、淘宝、当当网
读者须知：本书配套示例源码、PPT课件、教学视频与作者答疑服务，购买之后可加粉丝群

本书封面

在这里插入图片描述

第11章 Spark SQL自定义函数

本章将带你深入了解Spark中的用户自定义函数（User Defined Function，UDF）与自定义聚合函数（User-Defined Aggregate Functions，UDAF），探索它们在数据处理中的独特魅力。通过学习和实践，你将能够灵活运用这些高级功能，满足复杂的数据处理需求。本章是你掌握Spark高级编程的必读篇章。
本章主要知识点：

用户自定义函数UDF
用户自定义聚合函数UDAF

11.1 用户自定义函数UDF

Spark用户自定义函数是一种功能强大的工具，它赋予用户根据特定需求扩展Spark功能的能力。本节将详细阐述UDF的概念及其特性，并通过三个实战案例深入浅出地展示UDF的便捷性和实用性，帮助读者更好地理解和应用这一功能。

11.1.1 UDF函数的概念及其特点

定义与用途

定义：UDF是用户定义的函数，用于将一列或多列数据作为输入，并生成一个新的列作为输出。
用途：UDF在Spark中主要用于数据清洗和转换、特征工程、数据验证和过滤等场景。通过自定义函数，用户可以对数据进行各种复杂的处理，如字符串处理、日期格式转换、生成新的特征列等。

特点与优势

灵活性：UDF提供了用户自定义数据转换的能力，使得处理复杂的数据变得更加简单。
可复用性：一旦定义了UDF，就可以在不同的数据集上重复使用，提高了代码的可复用性。
可扩展性：UDF可以与Spark集群无缝集成，能够处理大规模数据集。

创建与使用

创建UDF：在Spark SQL中，可以通过spark.udf.register方法注册UDF。首先，需要定义一个普通的Scala函数，然后将这个函数注册为UDF。注册时，需要指定UDF的名称和处理逻辑。也可以通过spark.udf.register方法注册UDF，这种方法通常用于在DataFrame API中使用。
综上所述：Spark SQL在SQL查询中使用UDF时，可以直接在SELECT语句中调用注册的UDF名称。在DataFrame API中使用UDF时，可以通过withColumn或select方法将UDF应用于DataFrame的列上，并生成新的列。
接下来通过3个UDF案例带领读者使用UDF函数。

11.1.2 UDF案例1：字符串处理

本案例对字符串进行处理，在name字段的所有数据前面加上“Name:”。
文件名称为11student.json，JSON文件内容如下：

{"name":"A","lesson":"Math","score":100}
{"name":"B","lesson":"Math","score":100}
{"name":"C","lesson":"Math","score":99}
{"name":"D","lesson":"Math","score":98}
{"name":"A","lesson":"E","score":100}
{"name":"B","lesson":"E","score":99}
{"name":"C","lesson":"E","score":99}
{"name":"D","lesson":"E","score":98}

字符串处理的示例代码如代码11-1所示。
代码11-1 UDF01.scala

package chapter11
import org.apache.spark.sql.SparkSession

object UDF01 {

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

    val spark: SparkSession = SparkSession
      .builder()
      .appName("")
      .master("local[*]")
      .getOrCreate()

    val df = spark.read.json("BookData/input/10student.json")

    df.show()

    spark.udf.register("addName", (x: String) => "Name:" + x)

    df.createOrReplaceTempView("people")

    spark.sql("Select addName(name), score from people").show()

  }
}

执行以上代码，输出结果如图11-1所示。
在这里插入图片描述

11.1.3 UDF案例2：GEOHASH算法

本案例要求如下：
（1）通过Spark SQL处理一个包含用户登录信息的CSV文件。
（2）通过自定义用户定义函数（UDF）来计算每个登录记录的地理位置信息的GeoHash 编码。
（3）通过GeoHash编码得到地理位置。在编程中，GeoHash编码是一种将地理位置（经纬度）转换为紧凑字符串的方法，主要用于地理数据的存储和查询。GeoHash编码通过将二维地理空间分割成一系列正方形的网格，并将这些网格映射到二进制字符串上来实现。
本案例的数据包括地理位置数据和用户登录数据。
地理位置数据的文件名称为11area_data.txt，文件内容如下：

广西壮族自治区 河池市  凤山县  24.560064974995992 107.01971572194900  
广西壮族自治区 河池市  巴马瑶族自治县  24.15759554873569  107.20766596976145
陕西省 渭南市  蒲城县  34.96769665054483  109.62824611949353
河南省 开封市  顺河回族区  34.81777146999021  114.42852744048099
河南省 周口市  太康县  34.09709624887362  114.85570075513604
河南省 三门峡市 湖滨区  34.77177767294712  111.28129514586448
辽宁省 阜新市  彰武县  42.52375443552578  122.47417316389497
吉林省 吉林市  磐石市  43.0574561133298 126.17462779101467
广东省 湛江市  雷州市  20.796584309563634 110.01263612715134
甘肃省 白银市  白银区  36.501821828710106 104.20564932849953

用户登录数据的文件名称为11user_login.txt，文件内容如下：

uid01,login,2023-11-30,24.560064974995992,107.01971572194807
uid02,login,2023-11-30,34.96769665054483,109.62824611949353
uid03,login,2023-11-30,36.501821828710106,104.20564932849953

首先在pom.xml中添加以下依赖：

<dependency>
    <groupId>ch.hsr</groupId>
    <artifactId>geohash</artifactId>
    <version>1.4.0</version>
</dependency>

GEOHASH算法的示例代码如代码11-2所示。
代码11-2 UDF02.scala

package chapter11

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DataTypes, StructType}

/**
 * author: yuhui
 * descriptions:
 * 1）通过Spark SQL处理一个包含用户登录信息的CSV文件
 * 2）通过自定义用户定义函数（UDF）来计算每个登录记录的地理位置信息的GeoHash编码
 * 3）通过GeoHash编码得到地理位置
 * date: 2024 - 11 - 28 3:45 下午
 */
object UDF02 {

  def main(args: Array[String]): Unit = {
    val session = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getSimpleName)
      .getOrCreate()

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

    val schema = new StructType()
      .add("uid", DataTypes.StringType)
      .add("login", DataTypes.StringType)
      .add("dt", DataTypes.StringType)
      .add("lat", DataTypes.DoubleType)
      .add("lng", DataTypes.DoubleType)

    val frame = session.read
      .schema(schema)
      .csv("BookData/input/11user_login.txt")

    frame.createTempView("tb_log")

    val schema2 = new StructType()
      .add("province", DataTypes.StringType)
      .add("city", DataTypes.StringType)
      .add("district", DataTypes.StringType)
      .add("lat", DataTypes.DoubleType)
      .add("lng", DataTypes.DoubleType)

    val frame2 = session.read
      .schema(schema2)
      .csv("BookData/input/11area_data.txt")

    frame2.createTempView("tb_area")

    /**
     * 自定义函数UDF逐行计算函数
     * 1 定义一个scala函数
     * 2 注册到spark-sql中
     */
    val f = (lat: Double, lng: Double) => {
      GeoHash.withCharacterPrecision(lat, lng, 6).toBase32
    }

    // 注册
    session.udf.register("my_geo", f)
    session.sql(
      """
        |WITH aa AS (
        |    SELECT
        |        uid,
        |        login,
        |        dt,
        |        my_geo(lat, lng) AS geo_str_log
        |    FROM
        |        tb_log
        |),
        |bb AS (
        |    SELECT
        |        province,
        |        city,
        |        district,
        |        my_geo(lat, lng) AS geo_str_area
        |    FROM
        |        tb_area
        |)
        |SELECT
        |    aa.uid,
        |    aa.login,
        |    aa.dt,
        |    bb.province,
        |    bb.city,
        |    bb.district
        |FROM
        |    aa
        |LEFT JOIN
        |    bb
        |ON
        |    aa.geo_str_log = bb.geo_str_area;
        |""".stripMargin).show()
  }
}

执行以上代码，输出结果如图11-2所示。
在这里插入图片描述

11.1.4 UDF案例3：余弦相似度算法

本案例要求如下：
（1）通过一个人的年龄、身高、体重、颜值、分值作为一个数组，形成特征。
（2）使用余弦相似度函数计算出相似度。
余弦相似度公式如图11-3所示。
在这里插入图片描述
这里的余弦相似度算法是一种常用的度量两个非零向量之间相似度的方法。以下是对该算法的原理解说：
余弦相似度算法通过计算两个向量夹角间的余弦值来衡量两个个体之间的相似程度。在向量空间中，两个向量的夹角余弦值越接近1，表明两个向量的夹角越小，即它们之间的相似度越高；反之，余弦值越接近-1，表示两个向量的方向差异越大，即相似度越低；接近0则表示两个向量在方向上几乎无关。
数据文件名称为11features.csv，文件内容如下：

id,name,age,height,weight,yanzhi,score
1,a,18,172,120,98,68.8
2,b,28,175,120,97,68.8
3,c,30,180,130,94,88.8
4,d,18,168,110,98,68.8
5,e,26,165,120,98,68.8
6,f,27,182,135,95,89.8
7,g,19,171,122,99,68.8

余弦相似度算法的示例代码如代码11-3所示。
代码11-3 UDF03.scala

package chapter11

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.SparkSession

import scala.collection.mutable.WrappedArray

/**
 * author: yuhui
 * descriptions:
 * 1）通过一个人的age、height、weight、yanzhi、score作为特征
 * 2）使用余弦相似度函数计算出相似度
 * date: 2024 - 11 - 28 3:45 下午
 */
object UDF03 {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    // 加载数据
    val df = spark.read.option("inferSchema", true).option("header", true).csv("BookData/input/11features.csv")
    df.show()

    // 将数据转换为包含特征数组的DataFrame
    val featuresDF = df.selectExpr("id", "name", "array(age,height,weight,yanzhi,score) as features")

    // 将表自己和自己join，得到每个人和其他所有人的连接行
    val joined = featuresDF.as("a").join(featuresDF.as("b"), $"a.id" < $"b.id")

    // 定义一个计算余弦相似度的函数
    val cosinSim: UserDefinedFunction = udf((f1: WrappedArray[Double], f2: WrappedArray[Double]) => {
      val norm1 = Math.sqrt(f1.map(Math.pow(_, 2)).sum)
      val norm2 = Math.sqrt(f2.map(Math.pow(_, 2)).sum)
      val dotProduct = f1.zip(f2).map(p => p._1 * p._2).sum
      dotProduct / (norm1 * norm2)
    })

    // 注册UDF
    spark.udf.register("cos_sim", cosinSim)

    // 创建临时视图并计算余弦相似度
    joined.select($"a.id".as("aid"), $"a.features".as("afeatures"),
      $"b.id".as("bid"), $"b.features".as("bfeatures"))
      .createTempView("joined_temp")

    // 使用UDF直接在DataFrame SQL中计算余弦相似度
    spark.sql("SELECT aid, bid, cos_sim(afeatures, bfeatures) AS cos_sim_result FROM joined_temp").show()

    // 使用UDF直接在DataFrame API中计算余弦相似度
    joined.select($"a.id".as("aid"), $"b.id".as("bid"), cosinSim($"a.features", $"b.features").as("cos_sim_result")).show()

    // 关闭Spark
    spark.close()
  }
}

执行以上代码，输出结果如图11-4所示。
在这里插入图片描述

11.1.5 UDF注意事项

UDF在Spark SQL中通常比内置函数或DataFrame/Dataset API中的操作要慢，因为UDF会逐个元素处理数据，并且可能无法利用Spark的优化引擎（如Catalyst优化器）。
尽量使用DataFrame/Dataset API中的内置函数和转换操作，因为它们通常更高效且易于优化。
当确实需要自定义逻辑时，才考虑使用UDF。在定义UDF时，确保为UDF指定了正确的返回类型，以便Spark可以正确地处理数据。

11.2 用户自定义聚合函数UDAF

在Spark SQL中，用户自定义聚合函数（UDAF）允许用户实现复杂的聚合逻辑，这些逻辑不能通过Spark SQL的内置聚合函数（如sum()、avg()、max()、min()等）直接实现。UDAF是针对整组数据行执行的，而不是单独的数据行，并且可以在多个阶段（如初始化、累积和合并）执行自定义逻辑。
然而，需要注意的是，在Spark 2.x及更新版本中，官方并没有直接提供创建UDAF的API，与Spark 1.x中的Aggregator或UserDefinedAggregateFunction接口相比，用户更多地倾向于使用Aggregator或自定义Dataset转换来实现类似的功能。
不过，对于Spark SQL来说，用户仍然可以通过编写Scala/Java代码使用UserDefinedAggregateFunction接口来创建UDAF，并注册为SQL函数，以便在SQL查询中使用。
聚合函数UDAF分为弱类型聚合函数类和强类型聚合函数类。创建弱类型聚合函数类需要继承UserDefinedAggregateFunction，创建强类型聚合函数类需要继承Aggregator[输入类型,缓冲区类型,输出类型]）。下面给出UDAF的编程模板。

11.2.1 UDAF的编程模板

代码11-4将演示UADF的编程模板，模板中有8个方法需要实现，读者可以根据注释来理解这8个函数。
代码11-4 MyAvgUDAF.scala

package chapter11

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, StructType}

object MyAvgUDAF extends UserDefinedAggregateFunction {

  // 函数输入的字段Schema（字段名-字段类型）
  override def inputSchema: StructType = ???

  // 聚合过程中，用于存储局部聚合结果的Schema
  // 比如求平均薪资，中间缓存（局部数据薪资总和，局部数据人数总和）
  override def bufferSchema: StructType = ???

  // 函数的最终返回结果数据类型
  override def dataType: DataType = ???

  // 这个函数是不是稳定一致的？（对一组相同的输入，永远返回相同的结果），只要是确定的，就写true
  override def deterministic: Boolean = true

  // 对局部聚合缓存的初始化方法
  override def initialize(buffer: MutableAggregationBuffer): Unit = ???

  // 聚合逻辑所在方法，框架会不断地传入一个新的输入Row来更新你的聚合缓存数据
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = ???

  // 全局聚合：将多个局部缓存中的数据聚合成一个缓存
  // 比如薪资和薪资累加、人数和人数累加
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = ???

  // 最终输出
  // 比如从全局缓存中获取薪资总和/人数总和
  override def evaluate(buffer: Row): Any = ???
}

11.2.2 UDAF原理讲解

UDAF原理（见图11-5）是将数据通过聚合方式得出结果，聚合过程是分步骤进行的。
下面对聚合过程中的每一个步骤进行描述。
（1）通过两个Task分别读取数据，并将其转换为Row类型数据。
（2）通过update方法将每一行输入数据不断向buffer中聚合，本例中的聚合逻辑是将薪资和人数按照数据条数不断累计（+1）。
（3）通过buffer方法分别对薪资和人数总数进行累积。
（4）通过merge方法将属于同一个group的多个局部聚合buffer中的结果再次聚合并保存到全局聚合buffer中。
（5）通过buffer方法将全部聚合的缓存进行累加。
（6）通过evaluate方法输出最终结果。
聚合过程主要包括局部聚合和全局聚合两个阶段。

局部聚合（Update）：结果保存在一个局部buffer中。
全局聚合（Merge）：将多个局部buffer再聚合成一个全局buffer。
最终运算（Evaluate）通过evaluate对全局聚合后的buffer中的数据进行运算，得出所需的结果。

11.2.3 弱类型用户自定义聚合函数

本小节将使用弱类型用户自定义UDAF来求薪资的平均值。程序所需数据和字段如下，这些数据可以存储到List集合中。

+------------------+------+
|                 name|salary|
+------------------+------+
|                余辉 | 20000|
|  视频号：辉哥大数据 | 30000|
|  抖音：辉哥大数据   | 40000|
+------------------+------+

使用弱类型用户自定义UDAF求薪资的平均值的示例代码如代码11-5所示。
代码11-5 UDAFWeak.scala

package chapter11

import org.apache.spark.SparkConf
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, DataTypes, DoubleType, LongType, StructField, StructType}

/**
 * 使用弱类型用户自定义UDAF入门示例：求薪资的平均值
 */
class UDAFWeak extends UserDefinedAggregateFunction {
  // 函数输入的数据结构，需要new一个具体的结构对象，然后添加结构
  override def inputSchema: StructType = {
    new StructType().add("salary",LongType)
  }

  // 计算时的数据结构
  override def bufferSchema: StructType = {
    new StructType().add("sum",LongType).add("conut",LongType)
  }

  // 函数返回的数据类型
  override def dataType: DataType = DoubleType

  // 表述函数是否稳定
  override def deterministic: Boolean = true

  // 表述的是函数计算之前的缓冲区的初始化。buffer(0)表示第一个结构：sum， buffer(1)表示第二个结构：count
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }

  // 根据查询结构来更新缓冲区数据sum + = input.getLong  count+=1
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }

  // 将多个节点的缓冲区合并
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  // 计算
  override def evaluate(buffer: Row): Any = {
    buffer.getLong(0).toDouble / buffer.getLong(1)
  }
}

object UDAFWeak{
  def main(args: Array[String]): Unit = {
    // 创建配置对象
    val conf = new SparkConf().setAppName("Spark01_Custom"). setMaster("local[*]")
    val spark = SparkSession.builder().config(conf).getOrCreate()
    // 隐式转换（RDD转换DF/DS需要引入隐式转换）
    import spark.implicits._
    val frame =  spark.sparkContext.parallelize(
      List(
        ("余辉",20000),
        ("视频号：辉哥大数据",30000),
        ("抖音：辉哥大数据",40000))
    ).toDF("name","salary")
    // 创建全局视图
    frame.createGlobalTempView("people")
    frame.show()

    // 创建聚合函数对象
    val udaf = new UDAFWeak
    // 注册聚合函数
    spark.udf.register("avgSalary",udaf)
    // frame.select("salary").show()
    // sql  这里表名要把全局名也写上
    spark.sql("select avgSalary(salary) from global_temp.people").show
  }
}

运行以上程序，输出结果如下：

+-----------------+
|avgsalary(salary)|
+-----------------+
|            30000.0|
+-----------------+

11.2.4 强类型用户自定义聚合函数
本小节将使用强类型用户自定义聚合函数（UDAF）求薪资的平均值。程序所需数据和字段如下，这些数据可以存储到List集合中。

+------------------+------+
|                 name|salary|
+------------------+------+
|                余辉 | 20000|
|  视频号：辉哥大数据 | 30000|
|  抖音：辉哥大数据   | 40000|
+------------------+------+

使用强类型用户自定义聚合函数求薪资的平均值的示例代码如代码11-6所示。
代码11-6 UDAFStrong.scala

package chapter11

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.Encoder
import org.apache.spark.sql.Encoders
import org.apache.spark.sql.SparkSession

/**
 *  使用强类型用户自定义UDAF入门示例：求薪资的平均值
 */

// 既然是强类型，可能有case类
case class Employee(name: String, salary: Long)

case class Average(var sum: Long, var count: Long)

object UDAFStrong extends Aggregator[Employee, Average, Double] {
  // 定义一个数据结构，包括工资总数和工资总个数，初始都为0
  def zero: Average = Average(0L, 0L)

  // Combine two values to produce a new value. For performance, the function may modify 'buffer'
  // and return it instead of constructing a new object
  def reduce(buffer: Average, employee: Employee): Average = {
    buffer.sum += employee.salary
    buffer.count += 1
    buffer
  }

  // 聚合不同execute的结果
  def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count
    b1
  }

  // 计算输出
  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count

  // Encoders.product是进行Scala元组和case类转换的编码器
  def bufferEncoder: Encoder[Average] = Encoders.product

  // 设定最终输出值的编码器
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

object UDAFStrongMain {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val ds =  spark.sparkContext.parallelize(
      List(
        Employee("余辉",20000),
        Employee("视频号：辉哥大数据",30000),
        Employee("抖音：辉哥大数据",40000))
    ).toDS()
    ds.show()

    // Convert the function to a 'TypedColumn' and give it a name
    val averageSalary = UDAFStrong.toColumn.name("avgsalary")
    val result = ds.select(averageSalary)
    result.show()
  }
}

执行以上代码，输出结果如下：

+--------------+
| avgsalary     |
+--------------+
|        30000.0|
+--------------+

11.2.5 UDAF注意事项

Spark SQL从2.x版本开始更多地推荐使用Aggregator接口。不过，由于Aggregator通常与Dataset API一起使用，而UDAF更常用于SQL查询中。因此，如果需要在SQL查询中使用自定义聚合，可能需要定义一个包装了Aggregator的UDF，如前面示例代码中用到的stringConcatUDAF.toColumn（尽管这不是直接注册的UDAF，而是利用Aggregator实现的一个UDF的近似）。
直接注册UserDefinedAggregateFunction作为SQL函数在Spark SQL中并不是非常直观，并且通常不如使用Dataset API灵活。如果你需要在SQL查询中使用复杂的聚合逻辑，可以考虑使用视图或子查询结合内置的SQL函数来达成目的，或者将数据处理逻辑移到DataFrame/Dataset API中。

11.3 本章小结

本章聚焦于Spark SQL中的自定义函数，详细讲解了用户自定义函数（UDF）和用户自定义聚合函数（UDAF）的创建和使用。通过GEOHASH算法和余弦相似度算法两个具体案例，展示了UDF在数据处理中的强大功能，同时提醒读者在编写UDF时需要注意的事项，以确保函数使用的正确性和性能。在UDAF部分，本章提供了UDAF的编程模板，并分别介绍了弱类型和强类型UDAF的创建方法，让读者能够根据需要选择合适的UDAF类型。此外，还总结了UDAF在使用中需要注意的问题，帮助读者避免常见错误。通过本章的学习，读者将能够掌握Spark SQL自定义函数的编写和使用技巧，为数据处理提供更多灵活性和可能性。

本书其他章节

点赞+收藏+关注

在这里插入图片描述

码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

AI 已经能直接改设计稿了:Windows部署Penpot，接入Codex MCP实测

码道开发者社区

Agent 敢开写权限吗？——从「误改文件」到「凌晨3点推生产」，写权限的边界到底划在哪

分级、隔离、复核都做了，还是可能出意外。审计与回滚。全量操作日志：Agent 的每一次写操作都应该被记录。改了什么文件、执行了什么命令、调用了什么 API，全部留痕。这样出了问题，你能快速定位是哪个环节出的错。Claude Code 默认会在下记录会话日志，包括每一次工具调用。一键回滚：基于 Git 的自动 checkpoint。在 Agent 开始工作前，自动创建一个 commit 作为「安全点