通过spark从kafka到hdfs

Question

也许有人在一些项目中使用过:我在Spark中给Cassandra写信，在Spark中我使用kafkaUtils.createDirectStream。通过Spark-Cassandra连接器，我们可以使用Dstream.saveToCassandra方法。但对于保存/附加到hdfs，我使用：

stream.map(_.value).foreachRDD(rdd => {
  val conf = new Configuration()
  conf.set("fs.defaultFS", "hdfs://127.0.0.1:9000/")
  val fs = FileSystem.get(conf)
  fs.append(new Path("textfile.txt"))
    .write(rdd.collect().mkString.getBytes())
  fs.close()
})

但我不认为这是最好的方法。使用下面的代码可能会更好：

val prepStr = {
  val str = new StringBuilder
  if (!rdd.isEmpty()) {
    str.append(rdd.collect().mkString)
  }
  str
}

最后：

fs.append(path).write(prepStr.mkString.getBytes())

或者是有人用了另一种方式？