Spark Streaming - 使用从 Kafka 接收的元数据读取数据

Question

我对 Spark 世界非常陌生。我正在尝试为以下用例编写优化的解决方案：

1. 需要从 Kafka 读取流数据，主要是一些压缩文件的 S3 文件路径。
2. 从上面收到的文件路径中读取压缩文件并对其进行处理并将其存储回某个 S3 存储桶。

我能够阅读 Kafka 主题并获取文件路径，但不确定现在如何阅读此文件路径？ spark.read.binaryFile(filePath) 之类的东西。

任何帮助或指导将不胜感激。

Answer 1

这有很多examples

读取压缩文件

rdd = sc.textFile("s3://bucket/lahs/blahblah.*.gz")

---

没有你的代码很难，这里有一个关于读写的大纲

其余的从这个答案......

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

  //... create your schema
  // you path

  val filePath = "file:///tmp/spark/Blah/blah"
  // create it as your batch data
  // someBatchData

  // now read it, you need your schema and write it back in the process section below
  import spark.implicits._
  val kafkaStream = spark.readStream
    .format("kafka")
    .option("kafka.xyz.servers", "localhost:0001")
    .option("subscribe", "blah")
    .option("startingOffsets", "latest")
    .option("failOnDataLoss", "true") // stop and debug it
    .load()
    .as[String]

      kafkaStream.writeStream.foreachBatch((someBatchData:Dataset[String], batchId:Long) => {

    val records = someBatchData.collect()
    // go through all the records    
    records.foreach((path: String) => {
      val yourData = spark.read.schema(.. youfileSchema).json(..youPath)
      // write it back as you wanted..
      //
    })
  }).start()

  spark.streams.awaitAnyTermination()