当前位置:   article > 正文

利用Spark将Kafka数据流写入HDFS

利用Spark将Kafka数据流写入HDFS

利用Spark将Kafka数据流写入HDFS

在当今的大数据时代,实时数据处理和分析变得越来越重要。Apache Kafka作为一个分布式流处理平台,已经成为处理实时数据的事实标准。而Apache Spark则是一个强大的大数据处理框架,它提供了对数据进行复杂处理的能力。
本篇博客将介绍如何使用Spark来读取Kafka中的数据流,并将这些数据以CSV格式写入到HDFS中。
环境准备
在开始之前,确保你的开发环境中已经安装了以下软件:

Apache Kafka

#启动zookeeper
zkServer start
#启动kafka服务
kafka-server-start /opt/homebrew/etc/kafka/server.properties
  • 1
  • 2
  • 3
  • 4

Apache Spark

<properties>
      <scala.version>2.12.17</scala.version>
      <spark.version>3.0.0</spark.version>
 <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.12</artifactId>
      <version>${spark.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.12</artifactId>
      <version>${spark.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.12</artifactId>
      <version>${spark.version}</version>
    </dependency>
    <!-- Kafka Streaming dependency -->
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
      <version>${spark.version}</version>
    </dependency>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

Hadoop HDFS

#启动hdfs
start-dfs.sh
  • 1
  • 2

Java开发环境
此外,你需要在项目中包含Spark和Kafka的依赖库。

代码实现
首先,我们定义一个Scala case class Job 来表示从Kafka读取的每条记录的数据结构。

case class Job(
  Position: String,
  Company: String,
  Salary: String,
  Location: String,
  Experience: String,
  Education: String,
  Detail: String
)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

接下来,我们编写一个Kafka2Hdfs对象,并在其中实现main方法。这个方法将创建一个SparkSession,配置Kafka读取选项,并从Kafka中读取数据流。

object Kafka2Hdfs {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Kafka2Hdfs")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val kafkaOptions = Map[String, String](
      "kafka.bootstrap.servers" -> "127.0.0.1:9092",
      "subscribe" -> "flume",
      "startingOffsets" -> "earliest"
    )

    val stream = spark.readStream
      .format("kafka")
      .options(kafkaOptions)
      .load()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

我们使用subscribe选项指定Kafka中的topic名称,这里我们使用的是flume。startingOffsets选项设置为earliest,意味着我们从最早的记录开始读取数据。

接下来,我们将Kafka中的数据转换成DataFrame。我们首先将每条记录的value字段转换为字符串,然后使用map函数将每条记录解析为Job对象。

val jobDs = stream.selectExpr("CAST(value AS STRING)")
  .as[String]
  .map(line => {
    val fields = line.split(",")
    Job(
      Position = fields(0),
      Company = fields(1).trim,
      Salary = fields(2).trim,
      Location = fields(3).trim,
      Experience = fields(4).trim,
      Education = fields(5).trim,
      Detail = fields(6).trim
    )
  }).toDF()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

现在,我们已经有了一个包含Job对象的DataFrame。接下来,我们将这个DataFrame以CSV格式写入到HDFS中。我们使用writeStream方法,并设置format为csv,同时指定输出路径和检查点位置。

val query: StreamingQuery = jobDs.writeStream
  .format("csv")
  .option("header", "false")
  .option("path", "/")
  .option("checkpointLocation", "/ck")
  .start()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

注意,我们在这里将header选项设置为false,因为我们不打算在CSV文件中包含列名。path选项指定了输出文件的存储路径,而checkpointLocation选项指定了检查点的存储路径,这对于流处理的可靠性非常重要。

最后,我们调用awaitTermination方法来等待流处理的结束。在实际的生产环境中,你可能希望将这个流处理任务部署到一个集群上,并让它持续运行。

query.awaitTermination()
  • 1

总结
在这篇博客中,我们介绍了如何使用Spark读取Kafka中的数据流,并将这些数据以CSV格式写入到HDFS中。这种方法可以用于各种实时数据处理场景,例如日志分析、事件监控等。通过这种方式,我们可以将实时数据转换为静态数据,以便进行更深入的分析和处理。

完整代码:

package com.lhy.sparkkafka2hdfs

import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.sql.{DataFrame, Row, SparkSession}



case class Job(Position:String,Company:String,Salary:String,Location:String,Experience:String,Education:String,Detail:String)
object Kafka2Hdfs{
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("Kafka2Hdfs")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    val kafkaOptions = Map[String, String](
      "kafka.bootstrap.servers" -> "127.0.0.1:9092",
      "subscribe" -> "flume",
      "startingOffsets" -> "earliest"
    )

    val stream = spark.readStream
      .format("kafka")
      .options(kafkaOptions)
      .load()


    val jobDs = stream.selectExpr("CAST(value AS STRING)")
      .as[String]
      .map(line => {
        val fields = line.split(",")
        Job(
          Position = fields(0),
          Company = fields(1).trim,
          Salary = fields(2).trim,
          Location = fields(3).trim,
          Experience = fields(4).trim,
          Education = fields(5).trim,
          Detail = fields(6).trim
        )
      }).toDF()
//    val query = jobDs.writeStream.format("console").start()

    val query: StreamingQuery = jobDs.writeStream
      .format("csv")
      .option("header", "false")
      .option("path", "/")
      .option("checkpointLocation", "/ck")
      .start()

    query.awaitTermination()

  }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57

在这里插入图片描述
如有遇到问题可以找小编沟通交流哦。另外小编帮忙辅导大课作业,学生毕设等。不限于python,java,大数据,模型训练等。
在这里插入图片描述

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/413282
推荐阅读
相关标签
  

闽ICP备14008679号