PySpark - 加入两个 RDD - 无法加入 - 解包的值太多答案

【问题标题】：PySpark - Join two RDDs - Cannot join - Too many values to unpackPySpark - 加入两个 RDD - 无法加入 - 解包的值太多
【发布时间】：2021-02-12 10:56:31
【问题描述】：

我在 HDFS 中有两个文件（非常简单）：

测试：

1,Team1
2,Team2
3,Team3

测试2：

11,Player1,Team1
22,Player1,Team2
32,Player1,Team3

我想加入他们（通过 Team* 列）以获得以下输出：

Team1,1,11,Player1
Team3,3,32,Player1

为此，我使用以下代码：

test = sc.textFile("/user/cloudera/Tests/test")
test_filter = test.filter(lambda a: a.split(",")[1].upper() == "TEAM1" or a.split(",")[1].upper() == "TEAM2")
test_map = test_filter.map(lambda a: a.upper())
test_map = test_map.map(lambda a: (a.split(",")[1]))
for i in test_map.collect(): print(i)

test2=sc.textFile("/user/cloudera/Tests/test2")
test2_map = test2.map(lambda a: a.upper())
test2_map = test2_map.map(lambda a: (a.split(",")[2], a.split(",")[1]))
for i in test2_map.collect(): print(i)

test_join = test_map.join(test2_map)
for i in test_join.collect(): print(i)

但是当我尝试查看 join RDD 时，我收到以下错误：

  File "/usr/lib/spark/python/pyspark/rdd.py", line 1807, in <lambda>
    map_values_fn = lambda (k, v): (k, f(v))
ValueError: too many values to unpack

    at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:135)
    at org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:176)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:94)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)

我做错了什么？

谢谢！

【问题讨论】：

标签： apache-spark join pyspark rdd cloudera

【解决方案1】：

能否显示这两条语句的结果集： for i in test_map.collect(): print(i) & for i in test2_map.collect(): print(i)

你也可以试试下面的：

   test = sc.textFile("/user/cloudera/Tests/test")
   test_map = test.map(lambda a:a.upper())
   test_map = test_map.map(lambda a: (a.split(",")[1],a.split(",")[0]))
   for i in test_map.collect(): print(i)

   test2=sc.textFile("/user/cloudera/Tests/test2")
   test2_map = test2.map(lambda a: a.upper())
   test2_map = test2_map.map(lambda a: (a.split(",")[2], a.split(",")[1]))
   for i in test2_map.collect(): print(i)

   test_join = test_map.join(test2_map)
   for i in test_join.collect(): print(i)

【讨论】：