<think>我们使用Spark DataFrame的列操作来更新数据。在Python中，我们可以使用`withColumn`方法，结合Spark SQL的函数（例如`col`）来进行列操作。 步骤： 1. 导入必要的函数：`from pyspark.sql.functions import col` 2. 对empDF进行两次`withColumn`操作：一次更新age（age列加2），一次更新salary（salary列加500）。 3. 然后，我们按照3.2.3节的方法遍历并输出DataFrame中的所有数据。通常，遍历DataFrame的方法是使用`collect()`将数据收集到Driver端，然后逐行处理。 注意：由于我们修改列，如果列已经存在，`withColumn`会替换该列。 假设empDF的结构如下（根据第2题）： name: string, age: int, salary: long 我们将创建两个新的列（实际上是替换原有的列）： empDF = empDF.withColumn("age", col("age") + 2) empDF = empDF.withColumn("salary", col("salary") + 500) 然后，我们遍历输出每一行。在遍历时，我们可以通过行的字段名获取值。 代码示例： updated_empDF = empDF.withColumn("age", col("age") + 2).withColumn("salary", col("salary") + 500) rows = updated_empDF.collect() for row in rows: # 假设有"name", "age", "salary"三个字段 print(f"Name: {row['name']}, Age: {row['age']}, Salary: {row['salary']}") 但是注意：在Spark中，使用`collect()`会将所有数据拉到Driver端，如果数据量很大，可能会导致Driver内存溢出。但题目中只有三组数据，所以可以安全使用。 另外，根据引用[1]中的提示，我们之前可能是通过读取文本文件然后转换得到的DataFrame，但这里我们假设empDF已经正确创建，并且包含所需的列。 因此，我们编写如下代码： </think> ### Python实现：Spark DataFrame列操作与遍历输出 #### 1. **列值修改操作** 使用`withColumn()`方法结合Spark SQL函数进行列操作： ```python from pyspark.sql import functions as F # 修改列值操作 updated_empDF = empDF \ .withColumn("age", F.col("age") + 2) \ .withColumn("salary", F.col("salary") + 500) ``` #### 2. **遍历输出所有数据** 使用`collect()`收集数据并遍历输出（3.2.3节方法）： ```python # 收集数据并遍历输出 results = updated_empDF.collect() print("修改后的员工数据：") for row in results: print(f"Name: {row['name']}, Age: {row['age']}, Salary: {row['salary']}") ``` #### 3. **完整代码示例** ```python from pyspark.sql import SparkSession from pyspark.sql import functions as F # 创建SparkSession spark = SparkSession.builder.appName("DataFrameOperations").getOrCreate() # 创建示例DataFrame（假设empDF已按第2题创建） data = [ ("Michael", 30, 3000), ("Andy", 30, 4500), ("Justin", 28, 3500) ] columns = ["name", "age", "salary"] empDF = spark.createDataFrame(data, columns) # 列值修改操作 updated_empDF = empDF \ .withColumn("age", F.col("age") + 2) \ .withColumn("salary", F.col("salary") + 500) # 遍历输出结果 results = updated_empDF.collect() print("修改后的员工数据：") for row in results: print(f"{row['name']}, Age: {row['age']}, Salary: {row['salary']}") # 停止SparkSession spark.stop() ``` #### 输出结果示例： ``` 修改后的员工数据： Michael, Age: 32, Salary: 3500 Andy, Age: 32, Salary: 5000 Justin, Age: 30, Salary: 4000 ``` ### 关键原理说明 1. **列操作机制** `withColumn()`创建新列替换原列，`F.col("age") + 2`使用Spark的`Column`对象执行分布式计算，避免逐行处理[^1] 2. **类型安全处理** Spark自动处理数值类型转换（如`int`转`long`），无需显式类型声明 3. **遍历优化** `collect()`将分布式数据收集到Driver端，适合小数据集（如本例3条记录），大数据集应使用`foreachPartition()`[^1] 4. **数据不变性原则** 所有操作生成新DataFrame，原始`empDF`保持不变（Spark的惰性求值特性）[^1] ---