试图了解火花流

Question

我有这段代码：

val lines: org.apache.spark.streaming.dstream.InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, topics)
    lines.foreachRDD { rdd =>
      val df = cassandraSQLContext.read.json(rdd.map(x => x._2))
      sparkStreamingService.run(df)
    }
    ssc.start()
    ssc.awaitTermination()

我的理解是，foreachRDD 发生在驱动程序级别？所以基本上所有的代码块：

lines.foreachRDD { rdd =>
  val df = cassandraSQLContext.read.json(rdd.map(x => x._2))
  sparkStreamingService.run(df)
}

发生在驱动程序级别？ sparkStreamingService.run(df) 方法基本上对当前数据帧进行一些转换以产生一个新的数据帧，然后调用另一个方法（在另一个 jar 上）将数据帧存储到 cassandra。 因此，如果这一切都发生在驱动程序级别，我们没有使用 spark 执行器，我怎样才能使其并行使用执行器来并行处理 RDD 的每个分区

我的 spark 流服务运行方法：

    var metadataDataframe = df.select("customer", "tableName", "messageContent", "initialLoadRunning").collect()
 metadataDataframe.foreach(rowD => {
      metaData = populateMetaDataService.populateSiteMetaData(rowD)
      val headers = (rowD.getString(2).split(recordDelimiter)(0))

      val fields = headers.split("\u0001").map(
        fieldName => StructField(fieldName, StringType, nullable = true))
      val schema = StructType(fields)

      val listOfRawData = rowD.getString(2).indexOf(recordDelimiter)
      val dataWithoutHeaders = rowD.getString(2).substring(listOfRawData + 1)

      val rawData = sparkContext.parallelize(dataWithoutHeaders.split(recordDelimiter))
//      val rawData = dataWithoutHeaders.split(recordDelimiter)
      val rowRDD = rawData
        .map(_.split("\u0001"))
        .map(attributes => Row(attributes: _*))

      val newDF = cassandraSQLContext.createDataFrame(rowRDD, schema)
      dataFrameFilterService.processBasedOnOpType(metaData, newDF)
    })

Answer 1

foreachRDD 的调用确实发生在驱动程序节点上。但是，由于我们在 RDD 级别上运行，因此对其进行的任何转换都将是分布式的。在您的示例中，rdd.map 将导致每个分区被发送到特定的工作节点进行计算。

由于我们不知道您的 sparkStreamingService.run 方法在做什么，因此我们无法告诉您其执行的位置。

Answer 2

foreachRDD 可以在本地运行，但这只是意味着设置。 RDD本身就是一个分布式集合，所以实际工作是分布式的。

直接评论文档中的代码：

dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { record =>
    connection.send(record) // executed at the worker
  }
}

请注意，不基于 RDD 的代码部分在驱动程序处执行。它是使用 RDD 构建的代码，然后分发给工作人员。

您的代码具体注释如下：

   //df.select will be distributed, but collect will pull it all back in
   var metadataDataframe = df.select("customer", "tableName", "messageContent", "initialLoadRunning").collect()
 //Since collect created a local collection then this is done on the driver
 metadataDataframe.foreach(rowD => {
      metaData = populateMetaDataService.populateSiteMetaData(rowD)
      val headers = (rowD.getString(2).split(recordDelimiter)(0))

      val fields = headers.split("\u0001").map(
        fieldName => StructField(fieldName, StringType, nullable = true))
      val schema = StructType(fields)

      val listOfRawData = rowD.getString(2).indexOf(recordDelimiter)
      val dataWithoutHeaders = rowD.getString(2).substring(listOfRawData + 1)

      //This will run locally, creating a distributed record
      val rawData = sparkContext.parallelize(dataWithoutHeaders.split(recordDelimiter))
//      val rawData = dataWithoutHeaders.split(recordDelimiter)
      //This will redistribute the work
      val rowRDD = rawData
        .map(_.split("\u0001"))
        .map(attributes => Row(attributes: _*))
      //again, setting this up locally, to be run distributed
      val newDF = cassandraSQLContext.createDataFrame(rowRDD, schema)
      dataFrameFilterService.processBasedOnOpType(metaData, newDF)
    })

最终，您可能可以将其重写为不需要收集并将其全部分发，但 StackOverflow 不适合您