在Spark usign foreachPartition上发出HTTP post请求

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-http" % "10.2.4",
  "com.typesafe.akka" %% "akka-stream" % "2.6.14",
  "com.typesafe.akka" %% "akka-actor-typed" % "2.6.14"
)

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import scala.concurrent.Future
import scala.util.{Failure, Success}

def sendPostRequest(data: String, url: String)(implicit system: ActorSystem, mat: ActorMaterializer): Future[HttpResponse] = {
  val entity = HttpEntity(ContentTypes.`application/json`, data)
  val request = HttpRequest(method = HttpMethods.POST, uri = url, entity = entity)

  Http().singleRequest(request)
}

import org.apache.spark.sql.SparkSession
import scala.concurrent.ExecutionContext.Implicits.global

val spark = SparkSession.builder()
  .appName("Spark HTTP Post Example")
  .master("local[*]")
  .getOrCreate()

val data = spark.read.textFile("path/to/your/data.txt").rdd

data.foreachPartition { partition =>
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  partition.foreach { line =>
    val responseFuture: Future[HttpResponse] = sendPostRequest(line, "http://your-api-endpoint.com/post")

    responseFuture.onComplete {
      case Success(response) =>
        println(s"Response status: ${response.status}")
        response.discardEntityBytes()
      case Failure(exception) =>
        println(s"Request failed: ${exception.getMessage}")
    }
  }

  system.terminate()
}