从google pubsub到spark streaming的数据摄取速度很慢

Question

我正在使用google cloud Dataproc Spark集群来运行Spark streaming作业，该作业从多个PubSub订阅中读取数据并写入BigQuery。PubSub有500万个元素，滑动窗口为2分钟，批/窗口为30秒，我每批只能得到大约200,000个元素。我希望第一批就能全部拿到五百万。每个元素的大小约为140字节，并且是Avro消息格式。

我已经在数据流中实现了每秒100万个元素的速度，但我想用Dataproc做同样的事情。我尝试了Dataproc的自动缩放选项，也尝试了在数据流上工作的相同的波束管道代码。如果我增加订阅的数量，那么它可能会提供更多的吞吐量。是否有可能从单个订阅中获得1M个元素/秒的吞吐量？

以下是我的Scala代码：

// Reading from multiple PubSub.
for (a <- 0 to Integer.parseInt(subs)) {
  logger.info("SKCHECK : Creating stream : " + subscription + a)
  val everysub  = PubsubUtils.createStream(
      ssc, projectId, None, subscription + a,
      SparkGCPCredentials.builder.jsonServiceAccount(jsonPath).build(),
      StorageLevel.MEMORY_ONLY_SER).map(message => {
          // Method to send avro bytes message and get row
          val row : Row = avroMsgToRow(message.getData())
          row
      })
}

我的build.sbt看起来像这样：

    libraryDependencies ++= Seq(
      "org.apache.spark" %% "spark-core" % sparkVersion,
      "org.apache.spark" %% "spark-sql" % sparkVersion,
     // "org.apache.spark" %% "spark-mllib" % sparkVersion,
      "org.apache.spark" %% "spark-streaming" % sparkVersion,
     // "org.apache.spark" %% "spark-hive" % sparkVersion,
      "com.google.cloud" % "google-cloud-bigquery" % bigQueryVersion,
      "com.google.apis" % "google-api-services-bigquery" % googleApiBigQueryVersion,
      "com.google.cloud" % "google-cloud-nio" % gcsNioVersion,
      "com.sksamuel.avro4s" %% "avro4s-core" % avro4sVersion
    )

    // https://mvnrepository.com/artifact/com.google.cloud.bigdataoss/bigquery-connector
    libraryDependencies += "com.google.cloud.bigdataoss" % "bigquery-connector" % "0.10.0-hadoop2"

    // https://mvnrepository.com/artifact/com.spotify/spark-bigquery
    libraryDependencies += "com.spotify" %% "spark-bigquery" % "0.2.2"

    libraryDependencies += "com.google.apis" % "google-api-services-pubsub" % "v1-rev425-1.25.0"

    // https://mvnrepository.com/artifact/org.apache.bahir/spark-streaming-pubsub
    libraryDependencies += "org.apache.bahir" %% "spark-streaming-pubsub" % "2.3.0"


    // https://mvnrepository.com/artifact/org.scala-lang/scala-library
    libraryDependencies += "org.scala-lang" % "scala-library" % "2.10.0-M3"

    // https://mvnrepository.com/artifact/org.apache.spark/spark-avro
    libraryDependencies += "org.apache.spark" %% "spark-avro" % "2.4.0"

如果你需要更多的信息，请告诉我。

我希望通过单个PubSub订阅获得每秒100万个元素的数据摄取速度。

Answer 1

我认为你首先需要找出你的Spark流媒体工作的瓶颈。是CPU限制，内存限制，IO限制，还是因为Spark的一些参数导致它没有充分利用资源？我建议您从检查资源利用率开始，然后尝试不同的machine types。