DStream.foreachRDD 函数的含义是啥？

Posted 2023-03-23

【中文标题】DStream.foreachRDD 函数的含义是啥？

【英文标题】：What's the meaning of DStream.foreachRDD function?DStream.foreachRDD 函数的含义是什么？

【发布时间】：2016-07-25 02:44:25

【问题描述】：

在Spark Streaming中，每批数据间隔总是产生一个且只有一个RDD，为什么我们使用foreachRDD()来foreach RDD？ RDD只有一个，不需要foreach。 在我的测试中，我从来没有看到超过一个的 RDD。

【问题讨论】：

你到底真的明白答案了吗？

asyncified.io/2017/02/10/…

【参考方案1】：

DStream 或“离散流”是一种将连续数据流分解为小块的抽象。这称为“微批处理”。每个微批次成为一个 RDD，提供给 Spark 进行进一步处理。 每个批次间隔为每个 DStream 生成一个且只有一个 RDD。

RDD 是数据的分布式集合。将其视为一组指向实际数据在集群中的位置的指针。

DStream.foreachRDD 是 Spark Streaming 中的“输出运算符”。它允许您访问 DStream 的底层 RDD 以执行对数据执行实际操作的操作。例如，使用foreachRDD，您可以将数据写入数据库。

这里的小转折是理解 DStream 是一个有时间限制的集合。让我将其与经典集合进行对比：获取用户列表并对其应用 foreach：

val userList: List[User] = ???
userList.foreachuser => doSomeSideEffect(user)

这会将副作用函数doSomeSideEffect 应用于userList 集合的每个元素。

现在，假设我们现在不知道所有用户，所以我们无法建立他们的列表。取而代之的是，我们有大量用户，例如在早高峰时来到咖啡店的人：

val userDStream: DStream[User] = ???
userDstream.foreachRDDusersRDD => 
    usersRDD.foreachuser => serveCoffee(user)

注意：

DStream.foreachRDD 为您提供RDD[User]，

不是单个用户

。回到我们的咖啡示例，这是在某个时间间隔内到达的用户集合。 要访问集合的单个元素，我们需要对 RDD 进行进一步操作。在本例中，我使用 rdd.foreach 为每位用户提供咖啡。

考虑执行：我们可能有一群咖啡师在煮咖啡。那些是我们的执行者。 Spark Streaming 负责创建小批量用户（或订单），Spark 会将工作分配给咖啡师，以便我们可以并行化咖啡制作并加快咖啡供应。

【讨论】：

感谢您对分区/RDD 的更正。没有意识到这一点。

@maasg 我还是不明白！正如您所说，“每个批次间隔为每个 DStream 生成一个且只有一个 RDD”，它不需要foreachRDD() 的 API 来 foreach RDD，因为只有一个 RDD，foreachRDD()可以命名为getRDD() 并返回RDD 对象。你的例子可以这样修改： 　　val userDStream: DStream[User] = ??? 　　　userDstream.getRDD.foreachuser => serveCoffee(user)

@maasg 你的意思是使用 15 分钟作为时间间隔，9:00 生成一个 RDD，9:15 生成一个 RDD，9:30 生成一个 RDD，你可以使用 foreachRDD() foreach 三个RDD(contains 9:00, 9:15, 9:30)，但是假设现在是9:30，9:00和9:15的RDD是之前处理的，如果使用foreachRDD()，则只能处理一个9:30的RDD，9:30之前9:30之前的RDD怎么处理？ 9:30 之前的数据或 RDD 会在 9:30 消失，不是吗？在Spark steaming中，每个时间间隔处理当前批次间隔的数据，为什么要在9:30处理6:00的RDD，6:00的RDD应该在6:00处理。

@Guo，不，我不是这个意思。我想让你看到 foreachRDD 它是有时间限制的。您应该将foreachRDD 理解为scheduleOperationOnRddForEachTimeInterval(rdd => operation(rdd))。

@coolbreeze 就像在 Spark 中一样，如果没有任何操作，则不会实现转换。此外，如果我们没有在 Streaming Context 上注册输出操作（如 foreachRDD），当我们尝试启动它时会抛出异常：No output operations registered, so nothing to execute