我有一个分区Dataframe保存到hdfs中。我应该定期从Kafka主题加载新数据并更新hdfs数据。数据很简单：这只是一段时间内收到的推文数量。
所以，分区 Jan 18, 10 AM 可能有 2 ，我可能会收到Kafka发来的最新数据，包括3条推文，发在 Jan 18, 10 AM . 所以，我需要更新 Jan 18, 10 AM 价值 2+3=5 .
我现在的解决方案不好，因为我
将hdfs中的所有内容加载到ram中
从hdfs中删除所有内容
阅读Kafka的新Dataframe
合并2个Dataframe
将新的组合Dataframe写入hdfs。
（我在代码中为每个步骤提供了注解。）
问题是存储在hdfs上的Dataframe可能是1tb，这是不可行的。

import com.jayway.jsonpath.JsonPath
import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql.SparkSession
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}

//scalastyle:off
object TopicIngester {
  val mySchema = new StructType(Array(
    StructField("date", StringType, nullable = true),
    StructField("key", StringType, nullable = true),
    StructField("cnt", IntegerType, nullable = true)
  ))

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

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

    // read the old one
    val old = spark.read
      .schema(mySchema)
      .format("csv")
      .load("/user/maria_dev/test")

    // remove it
    val fs = FileSystem.get(new Configuration)
    val outPutPath = "/user/maria_dev/test"

    if (fs.exists(new Path(outPutPath))) {
      fs.delete(new Path(outPutPath), true)
    }

    // read the new one
    val _new = spark.read
      .format("kafka")
      .option("kafka.bootstrap.servers", "sandbox-hdp.hortonworks.com:6667")
      .option("subscribe", "test1")
      .option("startingOffsets", "earliest")
      .option("endingOffsets", "latest")
      .load()
      .selectExpr("CAST(value AS String)")
      .as[String]
      .map(value => transformIntoObject(value))
      .map(tweet => (tweet.tweet, tweet.key, tweet.date))
      .toDF("tweet", "key", "date")
      .groupBy("date", "key")
      .agg(count("*").alias("cnt"))

     // combine the old one with the new one and write to hdfs
    _new.union(old)
        .groupBy("date", "key")
        .agg(sum("sum").alias("cnt"))
        .write.partitionBy("date", "key")
        .csv("/user/maria_dev/test")

    spark.stop()
  }

  def transformIntoObject(tweet: String): TweetWithKeys = {
    val date = extractDate(tweet)
    val hashTags = extractHashtags(tweet)

    val tagString = String.join(",", hashTags)

    TweetWithKeys(tweet, date, tagString)
  }

  def extractHashtags(str: String): java.util.List[String] = {
    JsonPath.parse(str).read("$.entities.hashtags[*].text")
  }

  def extractDate(str: String): String = {
    JsonPath.parse(str).read("$.created_at")
  }

  final case class TweetWithKeys(tweet: String, date: String, key: String)

}

如何只加载必要的分区并更有效地更新它们？