基于条件过滤RDD并在Spark python中提取匹配数据答案

【问题标题】：Filtering RDD Based on condition and extracting matched data in Spark python基于条件过滤RDD并在Spark python中提取匹配数据
【发布时间】：2016-01-11 13:01:34
【问题描述】：

我有类似的数据，

cl_id      cn_id        cn_value
10004,     77173296     ,390.0
10004,     77173299     ,376.0
10004,     77173300     ,0.0
20005,     77173296     ,0.0
20005,     77173299     ,6.0
2005,      77438800     ,2.0

Cl_id ID：10004 ,20005

按 10004 过滤

10004,     77173296     ,390.0
10004,     77173299     ,376.0

按 20005 年过滤

20005,    77173296    ,0.0
20005,    77173299     ,6.0

现在我想要返回 RDD，

10004,cn_id,x1(77173296.value,77173300.value) ==> 10004,77173296,390.0,376.0
20005,cn_id,x1(77173296.value,77173300.value) ==> 20005,77173296,0.0,6.0

我想对这个return_RDD进行一些操作：

 def cal_for(rdd_list):
     #list.map(position1).filter(cn_id for this formula)-> calculate that formula -> store in a separate RDD -> Return that RDD

     rdd_list = rdd_list.map(lambda line:line.split(','))
     new_list = rdd_list.map(lambda x: (x[0]+', '+x[1],float(x[2])))
     new_list = rdd_list.filter(lambda x: x[1] == '77173296' && x[1] ==  '77173299')
     ## then  get the  RDD containing respective cn_values for cn_id 77173296 & cn_id 77173299
     ## and apply the following formula whre a=77173296.value b=77173299.value for cl_id 1004

    try:
        # want to process RDD with this  Formula
        return ((float(a)/float(a+b))*100)
    except ZeroDivisionError:
        return 0

#return or save cal_RDD

【问题讨论】：

标签： python apache-spark rdd

【解决方案1】：

无需两次过滤 RDD，修改和重新组合生成的 RDD，只需按 id 分组，然后映射值以进行所需的任何更改。如果您想根据某些条件进一步限制结果，请在映射时执行过滤器。

我真的不能给你更准确的答案：

a) 看起来您还没有真正尝试过实现这一点，并且 b) 我不完全确定你想要什么。

【讨论】：

因为我是新手，所以我会遇到这些问题。从输入 Rdd 我只想根据 cl_id 过滤结果并将处理后的 Rdd 传递给 cal_fo()
您可能需要阅读更多有关 Scala 和 Spark 的信息。例如，您不需要 return 关键字，并且一旦有了 RDD 就可以操作数据相对简单。我不完全确定您的问题是 Spark 问题，还是您希望有人为您编写逻辑。我认为您应该描述您希望实现的目标（不提及实际值）。
非常感谢史蒂文。我会在这方面做更多的工作，并尝试简化和实施它。非常感谢。