我在 pyspark/python 中有这段代码:
n --> 是新列表的个数
n = 2
new tuple ( x, y, z )
地点:
x --> index
y --> list [where variable n is a number of itens of list]
z --> value of front value of list
我正在执行这段代码:
RDD_Dados = RDD_one.map(lambda x: (x[0],
[list(x[i][1]) for i in range(n)],
x[i+1][1])
)
RDD_one的内容是:
(0, [(0, '5'), (1, '1'), (2, '2'), (3, '4'), (4, '4'), (5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(1, [(1, '1'), (2, '2'), (3, '4'), (4, '4'), (5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(2, [(2, '2'), (3, '4'), (4, '4'), (5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(3, [(3, '4'), (4, '4'), (5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(4, [(4, '4'), (5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(5, [(5, '3.5'), (6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(6, [(6, '-2'), (7, '1'), (8, '2'), (9, '0')])
(7, [(7, '1'), (8, '2'), (9, '0')])
(8, [(8, '2'), (9, '0')])
(9, [(9, '0')])
我想获取每个二元组的每个第二项并制作一个新列表，如下所示:
(0, ['5','1'], '2')
(1, ['1','2'], '4')
(2, ['2','4'], '4')
(3, ['4','4'], '3.5')
(4, ['4','3.5'], '-2')
(5, ['3.5','-2'], '1')
(6, ['-2','1'], '2')
(7, ['1','2'], '0')
但不起作用。
我一直收到这个错误:
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-12-3532f6a9c36a> in <module>(
59 RDD_Dados = RDD_one.map(lambda x: (x[0],[list(x[i][1]) for i in range(n)],x[i+1][1]))
File "C:\Spark\spark-2.3.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 229, in main
File "C:\Spark\spark-2.3.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 224, in process
File "C:\Spark\spark-2.3.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
vs = list(itertools.islice(iterator, batch))
File "<ipython-input-12-3532f6a9c36a>", line 59, in <lambda>
File "<ipython-input-12-3532f6a9c36a>", line 59, in <listcomp>
TypeError: 'int' object is not subscriptable
请您参考如下方法:
改变
RDD_Dados = RDD_one.map(lambda x: (x[0],
[list(x[i][1]) for i in range(n)],
x[i+1][1]))
对于
RDD_Dados = RDD_one.map(lambda x: (x[0],
[x[i][1] for i in range(1,n+1)],
x[n][1]))
range 从 0 开始，因此您在第一次迭代中执行 x[0][1]。但是，x[0] 是一个整数，因此是您的问题。另外，你不需要用 list 包裹 x[i][1] 否则你会得到例如(0, [ ['5'], ['1'] ], '2') 而不是 (0, ['5','1'], '2')
编辑:
对于更通用和更灵活的方法，您可以创建一个自定义函数，您可以在其中进一步扩展逻辑。
例如:
def my_logic(x):
ret = [x[0], []]
for i in range(1, n):
try:
ret[1].append(x[1][i][1])
except IndexError:
ret[1].append(0)
break
if len(x[1]) > n:
ret.append(x[1][n][1])
else:
ret.append(0)
return ret
然后就是做
RDD_Dados = RDD_one.map(my_logic)
