研究 RDD-pyspark 的不同元素

Posted 2023-04-15

【中文标题】研究 RDD-pyspark 的不同元素

【英文标题】：work on distinct elements of RDD-pyspark

【发布时间】：2016-01-18 14:25:34

【问题描述】：

我正在将来自 Kafka 的数据接收到 Spark Streaming 应用程序中。它采用 Transformed DStreams 的格式。然后我只保留我想要的功能。features=data.map(featurize) 这给了我 "name","age","whatever"。 然后我只想保留所有数据的名称features=data.map(featurize).map(lambda Names: Names["name"]

现在，当我打印此命令时，我得到了来自流应用程序的所有名称，但我想分别处理每个名称。 更具体地说，我想检查每个名称，如果我过去已经遇到过它，我想在它上面应用一个函数。否则我将继续我的申请。所以我希望每个名字都是一个字符串，这样我就可以将它插入到我的函数中，检查过去是否看到过一个字符串。

我知道 foreach 会给我每个 RDD ，但我仍然想分别处理 RDD 的每个名称。

pyspark 有什么办法吗？

【问题讨论】：

你试过updateStateByKey了吗，键是你的名字吗？

如何使用 updateStateByKey 来保留来自不同流窗口的名称？

您可以在这里找到有用的信息（找到 updateStateByKey 部分）：spark.apache.org/docs/latest/streaming-programming-guide.html

【参考方案1】：

所以我所做的是定义一个函数来检查我是否在过去看到过该名称，然后使用.filter(myfunc) 只使用我想要的名称...

现在的问题是，在每个新的流媒体窗口中，该功能都是从一开始就应用的，所以如果我在第一个窗口中看到约翰这个名字 7 次，我只会保留一次，但是如果我看到了约翰的名字在第二个窗口中出现了 5 次，我会再保留一次......

我想为所有流媒体应用程序保留一次 John 的名字... 对此有什么想法吗？