Update with inner join using spark dataframe/dataset/RDD

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// 创建SparkSession
val spark = SparkSession.builder()
  .appName("Update with inner join using Spark DataFrame")
  .getOrCreate()

// 加载需要更新的数据
val updateData = spark.read.format("csv").load("path_to_update_data.csv")

// 加载目标表数据
val targetTable = spark.read.format("csv").load("path_to_target_table.csv")

// 执行内连接操作
val joinedData = targetTable.join(updateData, targetTable("join_column") === updateData("join_column"), "inner")

// 添加需要更新的列
val updatedData = joinedData.withColumn("updated_column", when(col("condition"), col("new_value")).otherwise(col("old_value")))

// 将更新后的数据写入目标表
updatedData.write.format("csv").save("path_to_output_table.csv")