如何通过Spark streaming和Apache flume将Kafka主题加载到HDFS

agent.sources = kafka-source
agent.channels = memory-channel
agent.sinks = hdfs-sink

agent.sources.kafka-source.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafka-source.kafka.bootstrap.servers = <Kafka服务器地址>
agent.sources.kafka-source.kafka.topics = <Kafka主题名称>

agent.channels.memory-channel.type = memory

agent.sinks.hdfs-sink.type = hdfs
agent.sinks.hdfs-sink.hdfs.path = hdfs://<HDFS地址>/<目标路径>
agent.sinks.hdfs-sink.hdfs.fileType = DataStream
agent.sinks.hdfs-sink.hdfs.writeFormat = Text
agent.sinks.hdfs-sink.hdfs.rollInterval = 600
agent.sinks.hdfs-sink.hdfs.rollSize = 0
agent.sinks.hdfs-sink.hdfs.rollCount = 10000

agent.sources.kafka-source.channels = memory-channel
agent.sinks.hdfs-sink.channel = memory-channel

$ flume-ng agent -n agent -c conf -f <Flume配置文件路径> -Dflume.root.logger=INFO,console

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._

val ssc = new StreamingContext(sparkConf, Seconds(1))
val lines = ssc.textFileStream("<HDFS地址>/<目标路径>")
// 对数据进行处理
lines.foreachRDD { rdd =>
  // 处理每个RDD中的数据
  rdd.foreach { line =>
    // 处理每行数据
    println(line)
  }
}

ssc.start()
ssc.awaitTermination()

$ spark-submit --class <应用程序类名> --master <Spark主节点地址> <应用程序JAR包路径>