如何在Apache Spark 2.3中实现支持Append和Update模式的自定义Sink？

嘿，我来帮你拆解一下实现这个自定义Sink的核心要点，还有怎么搞定Append和Update这些不同的输出模式～

• 幂等性必须优先考虑：因为批处理/流处理很可能因为重试、故障重启导致同一个batchId被多次调用。你得在Sink里记录已经处理过的batchId，比如存在数据库表、本地文件或者分布式存储里，每次调用addBatch先检查这个批次是否已经处理过，避免重复写入数据。
• 资源要管好：比如连接数据库的话，别每次addBatch都新建连接，搞个连接池复用；操作文件的话，记得及时关闭文件句柄，不然容易资源泄漏。
• 数据类型要适配：Spark DataFrame的字段类型和你的目标存储（比如数据库、文件系统）类型得对应上，比如Spark的TimestampType要转成数据库的TIMESTAMP，DecimalType要匹配对应的精度，不然写入会报错。
• 错误处理要到位：写入失败时别直接抛异常就完事了，得捕获异常、记好日志，还可以加个重试机制（比如重试3次），或者配置告警通知，方便排查问题。
• 性能得优化：大数据量下别逐条写，一定要批量提交；如果是写入文件，可以按时间或者主键分区，提升写入和后续查询的效率；数据库的话，用批量插入/更新语句，别循环执行单条SQL。
• 配置要灵活：把目标地址、批量大小、连接池参数这些做成可配置的，别硬编码在Sink里，这样同一个Sink可以复用在不同场景。
• 状态要持久化：记录已处理的batchId别存在内存里，不然进程重启就没了，得存在持久化存储里，比如专门建个小表存batchId和处理状态。

首先得明确：Spark在不同输出模式下传给addBatch的DataFrame内容是不一样的，咱们得针对性处理：

Append模式 #

这种模式下，DataFrame里只有当前批次新增的数据，没有旧数据。处理起来相对简单：

• 直接把整个批次的数据批量写入目标存储就行，比如用数据库的批量插入语句，或者文件系统的追加写入。
• 重点还是保证幂等：用batchId做校验，避免重复写入同一个批次的数据。

Update模式 #

这种模式下，DataFrame里是当前批次中发生变化的行（包括新增和更新的行）。这时候需要根据主键来处理：

• 如果是写入关系型数据库：用UPSERT语句（比如MySQL的INSERT ... ON DUPLICATE KEY UPDATE，PostgreSQL的INSERT ... ON CONFLICT ... DO UPDATE）。先根据主键判断数据是否存在，存在就更新，不存在就插入。
• 如果是写入文件系统：比如用Delta Lake这类支持ACID的存储，直接用Merge操作把新数据合并到旧数据里；如果是普通文件，可能需要先读取对应分区的旧文件，合并新数据后再写入，但这种方式适合小数据量，大数据量还是推荐用支持Merge的存储。
• 同样要保证幂等：比如结合batchId和主键，避免同一个批次的更新操作重复执行（比如某个批次的更新已经执行过，就算再调用也不重复处理）。

给你个简化版的JDBC Sink例子，看看怎么结合模式处理：

import org.apache.spark.sql.DataFrame
import scala.collection.mutable

class JdbcSink(
  jdbcUrl: String,
  tableName: String,
  primaryKeys: Seq[String],
  outputMode: String
) extends Sink {
  // 模拟连接池（实际推荐用HikariCP这类成熟的连接池）
  private val connectionPool = new SimpleJdbcConnectionPool(jdbcUrl)
  // 持久化已处理的batchId，这里用内存集合示例，实际要存在数据库里
  private val processedBatches = mutable.Set[Long]()

  override def addBatch(batchId: Long, data: DataFrame): Unit = {
    // 先检查批次是否已处理，避免重复执行
    if (processedBatches.contains(batchId)) {
      println(s"Batch $batchId has been processed, skip.")
      return
    }

    // 按分区处理，提升性能
    data.foreachPartition { partition =>
      val conn = connectionPool.get()
      try {
        val sql = buildSqlStatement(data.columns)
        val stmt = conn.prepareStatement(sql)
        
        partition.foreach { row =>
          // 设置主键参数
          primaryKeys.zipWithIndex.foreach { case (key, idx) =>
            stmt.setObject(idx + 1, row.getAs(key))
          }
          // 设置其他字段参数
          data.columns.filter(!primaryKeys.contains(_)).zipWithIndex.foreach { case (col, idx) =>
            stmt.setObject(primaryKeys.size + idx + 1, row.getAs(col))
          }
          stmt.addBatch()
        }
        stmt.executeBatch()
      } finally {
        conn.close()
      }
    }

    // 标记批次为已处理
    processedBatches.add(batchId)
    // 实际要把processedBatches持久化到数据库，比如插入到batch_log表
  }

  // 根据输出模式构建SQL语句
  private def buildSqlStatement(columns: Array[String]): String = {
    val colsStr = columns.mkString(", ")
    val placeholders = columns.map(_ => "?").mkString(", ")
    
    outputMode match {
      case "append" => s"INSERT INTO $tableName ($colsStr) VALUES ($placeholders)"
      case "update" =>
        val updateCols = columns.filter(!primaryKeys.contains(_)).map(col => s"$col = VALUES($col)").mkString(", ")
        s"INSERT INTO $tableName ($colsStr) VALUES ($placeholders) ON DUPLICATE KEY UPDATE $updateCols"
    }
  }
}

// 简单的连接池实现（示例用，实际用成熟框架）
class SimpleJdbcConnectionPool(jdbcUrl: String) {
  def get(): java.sql.Connection = {
    java.sql.DriverManager.getConnection(jdbcUrl)
  }
}

内容的提问来源于stack exchange，提问作者user1870400