我会在有时间的时候完善这个答案。它来自我们的一个生产回购，它使用https://github.com/goibibo/dataplatform_utils 图书馆。

import com.goibibo.dp.utils.{KfUtils09, SparkKafkaUtils09}
import com.goibibo.dp.utils.SparkKafkaUtils09._
import org.apache.kafka.common.TopicPartition
import org.apache.spark.sql.{DataFrame, Row, SQLContext, SparkSession}
import org.apache.spark.streaming.kafka010.{KafkaUtils, LocationStrategies, OffsetRange}
import org.slf4j.{Logger, LoggerFactory}
import org.json4s.jackson.Serialization
import org.json4s._
import org.json4s.jackson.Serialization.{read, write}
import scala.collection.JavaConverters._

object KafkaToRedShift {

  val spark: SparkSession = getSparkSession("KafkaToRedShift") /* Implement this */
  val logger: Logger = LoggerFactory.getLogger(KafkaToRedShift.getClass)

  def createOffsetRange(kafkaBrokers: String, topics: Seq[String],
                        consumerGroup: String, maxMessagesPerPartition: Option[Int],
                        readFrom: String = READ_FROM_COMMITTED): (Seq[OffsetRange], KOffsets, Boolean) = {

    var isReadRequired = false
    val kafkaConfig = KfUtils09.createKafkaConfig(kafkaBrokers, consumerGroup)
    val topicsNames = topics.asJava
    val earliestOffsets: Map[TopicPartition, Long] = KfUtils09.getEarliestOffsets(topicsNames, kafkaConfig).get
    val latestOffsets: Map[TopicPartition, Long] = KfUtils09.getLatestOffsets(topicsNames, kafkaConfig).get
    val committedOffsets: Map[TopicPartition, Long] = KfUtils09.getCommittedOffsets(topicsNames, kafkaConfig).get
    val fromOffsets =
      if (READ_FROM_EARLIEST.equals(readFrom)) earliestOffsets
      else if (READ_FROM_LATEST.equals(readFrom)) latestOffsets
      else committedOffsets
    val offsetRanges: List[OffsetRange] = latestOffsets.toList.map(pairTopicPartitionAndOffset => {
      val (tp, untilOffset) = pairTopicPartitionAndOffset
      val totalMessagesInPartition = untilOffset - fromOffsets(tp)
      logger.info(s"${tp.topic} ${tp.partition} earliestOffsets =  $earliestOffsets committedOffsets = ${committedOffsets(tp)} fromOffsets = ${fromOffsets(tp)} untilOffset = $untilOffset")
      logger.info(s"${tp.topic} ${tp.partition} totalMessagesInPartition = $totalMessagesInPartition ")
      val newUntilOffset = if (maxMessagesPerPartition.isDefined) {
        if (totalMessagesInPartition > maxMessagesPerPartition.get) {
          logger.info(s"${tp.topic} ${tp.partition} totalMessagesInPartition = $totalMessagesInPartition higher than maxMessagesPerPartition = $maxMessagesPerPartition")
          val newUntilOffset = fromOffsets(tp) + maxMessagesPerPartition.get
          logger.info(s"${tp.topic} ${tp.partition} new untilOffset = $newUntilOffset")
          newUntilOffset
        } else {
          untilOffset
        }
      } else {
        untilOffset
      }
      if (newUntilOffset > fromOffsets(tp)) {
        isReadRequired = true
      }
      OffsetRange.create(tp.topic, tp.partition, fromOffsets(tp), newUntilOffset)
    })

    val latestOffsetsN = offsetRanges.map(o => (new TopicPartition(o.topic, o.partition), o.untilOffset)).toMap
    (offsetRanges, latestOffsetsN, isReadRequired)
  }

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

    implicit val formats: DefaultFormats.type =  org.json4s.DefaultFormats
    val kafkaBrokers: String = ConfigUtils.bootstrapServers
    val topics: Seq[String] = Seq(ConfigUtils.readTopic)
    val consumerGroup: String = ConfigUtils.kShiftGroup

    if (args.isEmpty) {
      throw new IllegalArgumentException("please provide filename")
    }

    val fileName = args.head
    logger.info(s"Found file name in argument: $fileName")
    val configStr = readFromFile(fileName) /* Implement this */
    val conf: MySqlConfig = Serialization.read[DatabaseConfig](configStr)

    val sql: String = conf.sql.get
    val fullTableName = conf.tableName.split('.')
    val tableSchema = fullTableName.head
    val table: String = fullTableName.tail.head
    val source = conf.source
    var batchpush = false
    val lockFileName = s"${topics}_file.lck"

    try {
      acquireLock(lockFileName) /* Implement this */
      implicit val formats = Serialization.formats(NoTypeHints)

      val (offsets, latestOffsets, isReadRequired) = createOffsetRange(kafkaBrokers, topics, consumerGroup, None)

      if (isReadRequired) {

        val fromOffset: Map[String, Map[String, Long]] = offsets.map { o =>
          (o.topic, o.topicPartition.partition(), o.fromOffset)
        }.groupBy(_._1).mapValues(value => value.map(t => (t._2.toString, t._3)).toMap)

        if(fromOffset.forall(o=> o._2.forall(t=> t._2==0))){
          batchpush = true
        }

        val fromOffsetStr: String = write(fromOffset)

        val toOffset: Map[String, Map[String, Long]] = offsets.map { o =>
          (o.topic, o.topicPartition.partition(), o.untilOffset)
        }.groupBy(_._1).mapValues(value => value.map(t => (t._2.toString, t._3)).toMap)

        val toOffsetStr: String = write(toOffset)

        val df: DataFrame = spark.
          read.
          format("kafka").
          option("kafka.bootstrap.servers", kafkaBrokers).
          option("subscribe", topics.mkString(",")).
          option("startingOffsets", fromOffsetStr).
          option("endingOffsets", toOffsetStr).
          load().cache

        df.createOrReplaceTempView("raw_data")

        val tfm = if(source == "mysql") {
          logger.info(s"Execute data transformation query: $sql")
          spark.sql(sql)
        } else {
          spark.sql(
            """
                with a as(
                    select cast(key as string) key, cast(value as string) value,
                            timestamp as ingestion_time,topic,partition,offset
                from raw_data
                ),
                b as
                (
                    select *,row_number() over(partition by key order by topic,partition,offset desc ) r from a
                )
                select * from b where r = 1
                """).drop("r").createOrReplaceTempView("dedup_data")
          // create or replace temp view casted_data as
          logger.info(s"Execute data transformation query: $sql")
          spark.sql(sql)
        }

        val columnNames = tfm.schema.map(_.name).map(c => s""" "$c" """).mkString(",")
        tfm.createOrReplaceTempView("tfm")

        val spectrumTableName = s"misc.${table}_realtime_tmp"
        spark.sql(s"drop table if exists $spectrumTableName")
        spark.table("tfm").repartition(3).write.format("parquet").mode("overwrite").saveAsTable(spectrumTableName)

        val redshiftSql: String =
          s"""
             |create temp table ${table}_realtime_staging
             |        distkey(key)
             |      sortkey(key)
             |      as
             |      select * from misc_e.${table}_realtime_tmp;
             |
             |      CREATE TABLE if not exists $tableSchema.${table}_realtime (like ${table}_realtime_staging);
             |
             |      delete from $tableSchema.${table}_realtime
             |      where key in (select key from ${table}_realtime_staging where key is not null);
             |
             |      insert into $tableSchema.${table}_realtime($columnNames)
             |      select $columnNames from ${table}_realtime_staging where raw_json != 'Deleted';
             |
             |      select 1;
         """.stripMargin

        logger.info(s"Excecuting Below Query in Redshift: $redshiftSql")
        RedshiftUtils.executeRedshiftSql(redshiftSql) /* Implement this */

        if (!SparkKafkaUtils09.commitOffsets(kafkaBrokers, consumerGroup, latestOffsets)) {
          logger.warn("Commit Offsets failed")
        } else {
          logger.info("Commit Offsets successful!")
        }

      }

    } catch {
      case e: Exception => {
        val stackTrace = org.apache.commons.lang.exception.ExceptionUtils.getStackTrace(e)
        logger.info(s"Releasing lock due to exception.. $stackTrace")
        releaseLock(lockFileName) /* Implement this */
        throw new Exception("exiting due to exception..", e)
      }

    } finally {
      //release the lock
      releaseLock(lockFileName) /* Implement this */
    }

    logger.info("All Done!!!")
    System.exit(0)
  }
}