如何在hortonworks中将spark流数据存储到hdfs？

8xiog9wr 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(342)

我用spark从Kafka主题中提取了数据流。这是我试过的代码。在这里，我只是在控制台中显示流数据。我想把这些数据作为文本文件存储在hdfs中。

import _root_.kafka.serializer.DefaultDecoder
import _root_.kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.storage.StorageLevel
object StreamingDataNew {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("Kafka").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
val kafkaConf = Map(
      "metadata.broker.list" -> "localhost:9092",
      "zookeeper.connect" -> "localhost:2181",
      "group.id" -> "kafka-streaming-example",
      "zookeeper.connection.timeout.ms" -> "200000"
    )
val lines = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](
      ssc,
      kafkaConf,
      Map("topic-one" -> 1), // subscripe to topic and partition 1
      StorageLevel.MEMORY_ONLY
    )
    println("printing" + lines.toString())
    val words = lines.flatMap { case (x, y) => y.split(" ") }
    words.print()

    ssc.start()
    ssc.awaitTermination()

  }
}

我发现我们可以使用“saveastextfiles”编写数据流，但是有人能清楚地提到如何使用上面的scala代码连接hortonworks并存储在hdfs中的步骤吗。

hdfs apache-kafka apache-spark spark-streaming hortonworks-sandbox

来源：https://stackoverflow.com/questions/50460451/how-to-store-spark-streaming-data-to-hdfs-in-hortonworks

1条答案

按热度按时间

k75qkfdt1#

我找到了答案，这个代码适合我。

package com.spark.streaming

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkContext
import org.apache.spark.sql._
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

object MessageStreaming {
  def main(args: Array[String]): Unit = {
    println("Message streaming")

    val conf = new org.apache.spark.SparkConf().setMaster("local[*]").setAppName("kafka-streaming")
    val context = new SparkContext(conf)
    val ssc = new StreamingContext(context, org.apache.spark.streaming.Seconds(10))
    val kafkaParams = Map(
      "bootstrap.servers" -> "kafka.kafka-cluster.com:9092",
      "group.id" -> "kafka-streaming-example",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "auto.offset.reset" -> "latest",
      "zookeeper.connection.timeout.ms" -> "200000"
    )
    val topics = Array("cdc-classic")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams))

    val content = stream.filter(x => x.value() != null)
    val sqlContext = new org.apache.spark.sql.SQLContext(context)
    import sqlContext.implicits._

    stream.map(_.value).foreachRDD(rdd => {

      rdd.foreach(println)
      if (!rdd.isEmpty()) {
     rdd.toDF("value").coalesce(1).write.mode(SaveMode.Append).json("hdfs://dev1a/user/hg5tv0/hadoop/MessagesFromKafka")

      }

    })
    ssc.start()
    ssc.awaitTermination

}}

赞(0）回复(0）举报 2021-06-07

我来回答

如何在hortonworks中将spark流数据存储到hdfs？

1条答案

相关问题

热门标签

最新问答