循环 PySpark 后的联合 RDD
Posted
技术标签:
【中文标题】循环 PySpark 后的联合 RDD【英文标题】：Union RDDs after a loop PySpark 【发布时间】：2016-11-10 17:02:49 【问题描述】：我正在使用 PySpark，我正在寻找一种修改 4 个 rdds 的方法，这些 rdds 包含在一个列表中。当我显示我的列表时，我有这样的东西：
所以给定：
for r in repartionned_rdd:
print r.collect()
给：
[(u'_guid_NCw7SuFnCh_mFW3SI3qTvBCbqXKD4mtsdJvWE7HNgNg=', (u'f', u'KSJakOd2|KtC9ZF9h'))]
[(u'_guid_OCs2au-sKnxzPE0uRPDP4hg1vvhgpzRAAYjNWRQpKbw=', (u'f', u'KxrylzuA|KpSXJwH2')), (u'_guid_txH15ULaeUDBC4Z_NlEOj2xoYBFa-08imqIBLfYsKps=', (u'f', u'bda54c71-cd1e-4eb7-856c-ba2e6def30c8|6e189e07-807e-41a2-a60a-b07d894a2905')), (u'_guid_ehCT6NyD9l3q3NV9ZroaWVEo3bnDt4tvbU_fMBrEn1g=', (u'm', u'537D69B4-743A-45B9-BED1-A25AA5926F13|2bb3e466-edc5-4302-b102-3bddb1f8c490|aa4760de-104c-4dc3-94c3-336427f89723')), (u'_guid_9F4Ph5GztLN9IlWNgZWKPMCcT4N3Je6-93iM_130F-c=', (u'f', u'KOQqBzhU|KrDt5GC4')), (u'_guid_nPlE_f-zoOHNYiXJSGXWoVryc1U4Bnfxkow3P0mDUFY=', (u'f', u'Kh3tIZR1|Khs0tRsh|K3geBqb_|KBrVNcDX|Jg2uDy8M|529816a3-ee43-4423-961f-8aedaf25d58c')), (u'331d8410d4924e72b0f0585e888c85ce', (u'f', u'1F37807A-CBEA-4B78-85D7-5A97B37B539E'))]
[(u'28b195c271f14a329235c262e7baecbf', (u'm', u'50c41480-a94e-4afa-a732-b6ed7a057239'))]
[(u'c65ac2064bc14116a363125392dcc6f7', (u'f', u'77e4b9b3-83b4-4553-b274-7a16f553cf05')), (u'171f92200d634d62bdc6685bdb7a94e3', (u'f', u'bdf53cb6-695d-4dde-b0c1-d1a34ebea6f7|a09e4074-c22e-48a1-9976-ee2151b5888c|K1Umlb5M|639B02B4-24AD-4069-99A2-C68E8C8F7F06|KjE3wXIr')), (u'_guid_wQZIzeFxciX9CIHUPeWOF2euOIC0jiosXVXN98_zCh8=', (u'f', u'F0992237-2598-4B13-AA8A-C37D436B901C|C80D1A89-DD84-4734-838F-128F99EBDD20|KthpuVu0')), (u'_guid_ufOcKO48drwr50yJN26NriX5MLYONwmALxWcmly7oqQ=', (u'f', u'KlY10YxX|KyCVx_km'))]
我的目标是为这个列表中的每个 rdd 添加一种新的“列”。此行将包含每个 rdd 的唯一索引。我的代码：
for i, rdd in enumerate(repartionned_rdd):
new_rdd = rdd.map(lambda x : x + (float(i), ))
print new_rdd.collect()
这给出了：
[(u'_guid_NCw7SuFnCh_mFW3SI3qTvBCbqXKD4mtsdJvWE7HNgNg=',
(u'f', u'KSJakOd2|KtC9ZF9h'), 0.0)]
[(u'_guid_OCs2au-sKnxzPE0uRPDP4hg1vvhgpzRAAYjNWRQpKbw=',
(u'f', u'KxrylzuA|KpSXJwH2'), 1.0),
(u'_guid_txH15ULaeUDBC4Z_NlEOj2xoYBFa-08imqIBLfYsKps=',
(u'f', u'bda54c71-cd1e-4eb7-856c-ba2e6def30c8|6e189e07-807e-41a2-a60a-b07d894a2905'), 1.0),
(u'_guid_ehCT6NyD9l3q3NV9ZroaWVEo3bnDt4tvbU_fMBrEn1g=',
(u'm', u'537D69B4-743A-45B9-BED1-A25AA5926F13|2bb3e466-edc5-4302-b102-3bddb1f8c490|aa4760de-104c-4dc3-94c3-336427f89723'), 1.0),
(u'_guid_9F4Ph5GztLN9IlWNgZWKPMCcT4N3Je6-93iM_130F-c=',
(u'f', u'KOQqBzhU|KrDt5GC4'), 1.0),
(u'_guid_nPlE_f-zoOHNYiXJSGXWoVryc1U4Bnfxkow3P0mDUFY=',
(u'f', u'Kh3tIZR1|Khs0tRsh|K3geBqb_|KBrVNcDX|Jg2uDy8M|529816a3-ee43-4423-961f-8aedaf25d58c'), 1.0),
(u'331d8410d4924e72b0f0585e888c85ce',
(u'f', u'1F37807A-CBEA-4B78-85D7-5A97B37B539E'), 1.0)]
[(u'28b195c271f14a329235c262e7baecbf',
(u'm', u'50c41480-a94e-4afa-a732-b6ed7a057239'), 2.0)]
[(u'c65ac2064bc14116a363125392dcc6f7',
(u'f', u'77e4b9b3-83b4-4553-b274-7a16f553cf05'), 3.0),
(u'171f92200d634d62bdc6685bdb7a94e3',
(u'f', u'bdf53cb6-695d-4dde-b0c1-d1a34ebea6f7|a09e4074-c22e-48a1-9976-ee2151b5888c|K1Umlb5M|639B02B4-24AD-4069-99A2-C68E8C8F7F06|KjE3wXIr'), 3.0),
(u'_guid_wQZIzeFxciX9CIHUPeWOF2euOIC0jiOsXVXN98_zCh8=',
(u'f', u'F0992237-2598-4B13-AA8A-C37D436B901C|C80D1A89-DD84-4734-838F-128F99EBDD20|KthpuVu0'), 3.0),
(u'_guid_ufOcKO48drwr50yJN26NriX5MLYONwmALxWcmly7oqQ=',
(u'f', u'KlY10YxX|KyCVx_km'), 3.0)]
所以我的 new_rdd 中的每一行都包含一个新列，具体来说就是 rdd 的索引（如代码中所述！）
我现在的目标是将所有这些新的 rdd 放在一个唯一的 rdd 中。 我试过这个：
all_rdds_list =[]
for i, rdd in enumerate(repartionned_rdd):
new_rdd = rdd.map(lambda x : x + (float(i), ))
all_rdds_list.append(new_rdd)
但是当我尝试显示我的 rdds 时，我得到了这个：
for x in all_rdds_list:
print x.collect()
结果：
[(u'_guid_NCw7SuFnCh_mFW3SI3qTvBCbqXKD4mtsdJvWE7HNgNg=',
(u'f', u'KSJakOd2|KtC9ZF9h'), 3.0)]
[(u'_guid_OCs2au-sKnxzPE0uRPDP4hg1vvhgpzRAAYjNWRQpKbw=',
(u'f', u'KxrylzuA|KpSXJwH2'), 3.0),
(u'_guid_txH15ULaeUDBC4Z_NlEOj2xoYBFa-08imqIBLfYsKps=',
(u'f', u'bda54c71-cd1e-4eb7-856c-ba2e6def30c8|6e189e07-807e-41a2-a60a-b07d894a2905'), 3.0),
(u'_guid_ehCT6NyD9l3q3NV9ZroaWVEo3bnDt4tvbU_fMBrEn1g=',
(u'm', u'537D69B4-743A-45B9-BED1-A25AA5926F13|2bb3e466-edc5-4302-b102-3bddb1f8c490|aa4760de-104c-4dc3-94c3-336427f89723'), 3.0),
(u'_guid_9F4Ph5GztLN9IlWNgZWKPMCcT4N3Je6-93iM_130F-c=',
(u'f', u'KOQqBzhU|KrDt5GC4'), 3.0),
(u'_guid_nPlE_f-zoOHNYiXJSGXWoVryc1U4Bnfxkow3P0mDUFY=',
(u'f', u'Kh3tIZR1|Khs0tRsh|K3geBqb_|KBrVNcDX|Jg2uDy8M|529816a3-ee43-4423-961f-8aedaf25d58c'), 3.0),
(u'331d8410d4924e72b0f0585e888c85ce',
(u'f', u'1F37807A-CBEA-4B78-85D7-5A97B37B539E'), 3.0)]
[(u'28b195c271f14a329235c262e7baecbf',
(u'm', u'50c41480-a94e-4afa-a732-b6ed7a057239'), 3.0)]
[(u'c65ac2064bc14116a363125392dcc6f7',
(u'f', u'77e4b9b3-83b4-4553-b274-7a16f553cf05'), 3.0),
(u'171f92200d634d62bdc6685bdb7a94e3',
(u'f', u'bdf53cb6-695d-4dde-b0c1-d1a34ebea6f7|a09e4074-c22e-48a1-9976-ee2151b5888c|K1Umlb5M|639B02B4-24AD-4069-99A2-C68E8C8F7F06|KjE3wXIr'), 3.0),
(u'_guid_wQZIzeFxciX9CIHUPeWOF2euOIC0jiOsXVXN98_zCh8=',
(u'f', u'F0992237-2598-4B13-AA8A-C37D436B901C|C80D1A89-DD84-4734-838F-128F99EBDD20|KthpuVu0'), 3.0),
(u'_guid_ufOcKO48drwr50yJN26NriX5MLYONwmALxWcmly7oqQ=',
(u'f', u'KlY10YxX|KyCVx_km'), 3.0)]
帮助？谢谢！
【问题讨论】：
【参考方案1】：你的方法有两个问题。首先，您使用在评估方法分配之前发生变化的变量。映射调用是转换，因此仅在您应用操作时执行（如collect）。这就是为什么当您在枚举循环中收集时会看到正确的附加列，但在后面的示例中，它会为每个映射选择 i 的最后一个值。
第二个问题是，如果您尝试将 rdds 合并在一起，您应该使用 union 函数而不是 rdds 列表。如果您确实想要一个 rdds 列表，那么您可以将下面的并集替换为您之前的列表追加。
full_rdd = None
for i, rdd in enumerate(repartionned_rdd):
new_rdd = rdd.map(lambda x : x + (float(i),))
if full_rdd is None:
full_rdd = new_rdd
else:
full_rdd = sc.union([full_rdd, new_rdd])
# This will force the lazy evaluation to execute now before `i` changes
full_rdd.count()
【讨论】：
Pyrce ：谢谢，但应用您的代码，我收到此错误消息：TypeError: union() 需要 2 个参数（给定 3 个） @DataAddicted opps，我错过了围绕 RDD 参数的列表。现在应该是正确的。【参考方案2】：最好的方法：
def get_population_id(repartionned_rdd):
idx = range(len(repartionned_rdd))
FullRDD = sc.emptyRDD()
for (i, rdd) in zip(idx, repartionned_rdd):
FullRDD = FullRDD.union(rdd.map(lambda x: x + (float(i),)))
return FullRDD
【讨论】：
以上是关于循环 PySpark 后的联合 RDD的主要内容，如果未能解决你的问题，请参考以下文章
PySpark：具有不同列的 DataFrames 的动态联合