从主RDD创建RDD

Question

我有一个RDD (RDD[(String，IterableEvent)]，它有一个键，表示一年中的一个月，值是该月发生的数百万个事件。

我想遍历每个键，并创建键事件的RDD。然后，我想为当月事件的每一天创建一个event RDD，这样我就可以将它们发送到相关的s3位置(“目录”结构是bucketName/year/month/ day )。

问题是，似乎你不能在另一个RDD的foreach中创建RDD。因此，我不确定如何在不将整个主RDD加载到内存中的情况下实现我想要的东西(这肯定会耗尽驱动程序的内存，并在一开始就失去使用Spark的意义)。

也许有一种方法可以使用Spark来实现我想要的东西，我只是对此并不了解，希望这里有人能帮上忙。

下面是我目前掌握的代码：

 private def store(
    eventsByMonth: RDD[(String, Iterable[Event])]
  )(
    implicit sqlContext: SQLContext
  ): Try[Unit] =
    Try(
      eventsByMonth
        .foreach {
          case (_, events: Iterable[Event]) =>
            writeToS3Files(sqlContext.sparkContext.parallelize(events.toSeq))
        }
    )

  private def writeToS3Files(events: RDD[Event])(
    implicit sqlContext: SQLContext
  ): Try[Unit] =
    Try(
      // outputFilePath will contain the day that these events are related to.
      events.groupBy(_.outputFilePath).foreach {
        case (filePath: String, eventsForFile: Iterable[Event]) =>
          writeToS3File(filePath, sqlContext.sparkContext.parallelize(eventsForFile.toSeq))
      }
    )

  private def writeToS3File(filePath: String, events: RDD[Event]): Try[Unit] = {
    val fileNameWithPath = s"${filePath}${UUID.randomUUID().toString}.gz"

    Try(events.saveAsTextFile(fileNameWithPath, classOf[GzipCodec]))
  }

Answer 1

我假设有一些方法可以确定事件发生的日期(例如，day (类型为Int)是事件的属性)。

您可以将RDD[(String，IterableEvent] )转换为一个PairRDD(K，V)，其中键(K)是事件发生的月份和日期，值(V)是在该月的某一天发生的所有事件。之后，您可以轻松地将数据转储到数据库中。

val eventsByMonthAndDate = eventsByMonth.flatMap { case (month, events) => events.map(e => ((month, e.day), e)) }
eventsByMonthAndDate.groupby(_._1).foreach(writeToDB)