Pyspark RDD“列表索引超出范围”错误答案

【问题标题】：Pyspark RDD "list index out of range" errorPyspark RDD“列表索引超出范围”错误
【发布时间】：2019-04-24 07:54:45
【问题描述】：

我有这种形式的 RDD：

[[['a'],['b,c,d','e,f,g']],[['h'],['i,j,k','l,m,n']]]

我想要达到的目标：

[['a ,b,c,d', 'a ,e,f,g'], ['h ,i,j,k', 'h ,l,m,n']]

我做了什么：

def pass_row(line):
  new_line = []
  key = ''.join(line[0])
  for el in line[1]:
    el = key +' ,'+ el
    new_line.append(el)
  return new_line

rdd.map(pass_row)

它适用于较小的数据样本。但是，当我尝试在我的整个数据集上运行它时，我在for el in line[1]: 上得到了list index out of range...

基本上我有一个键（比如说['a']），用于['b,c,d','e,f,g'] 中的~100 组不同的值。我的最终目标是以行的形式将其作为 spark 数据框：

col1 col2 col3 col 4
a     b    c    d
a     e    f    g
h     i    j    k
h     l    m    n

感谢您的建议！

【问题讨论】：

显然你有一个记录，它有一个元素而不是两个。所以，line[1] 出现错误。

标签： python python-3.x pyspark rdd

【解决方案1】：

您的错误似乎与您的数据比您的函数更相关（这似乎是正确的，尽管有点过于复杂），并且看起来您将它应用于没有line[1] 的行。

您能否确保line 的元素数量在您的实际数据集中是恒定的，例如：

def pass_row(line):
    assert len(line) == 2
    return [ "%s, %s" % (''.join(line[0]), el) for el in line[1]]

话虽如此，为了您的实际目标，您可能应该从那时起停止处理字符串并直接将数据作为二维数组获取，例如：

def pass_row(line):
    return [line[0] + el.split(',') for el in line[1]]

>>> a = [[['a'],['b,c,d','e,f,g']],[['h'],['i,j,k','l,m,n']]]
>>> b = [ pass_row(i) for i in a ]
>>> b
[[['a', 'b', 'c', 'd'], ['a', 'e', 'f', 'g']], [['h', 'i', 'j', 'k'], ['h', 'l', 'm', 'n']]]

在此警告，您不能直接使用该解决方案提供 DataFrame，因为每组前缀生成的行仍然嵌套在其自己的列表中（它只是一个“二维数组列表”）。

例如，使用 sum 函数，您可以轻松地在 reduce 步骤中转置：

>>> sum(b, [])
[['a', 'b', 'c', 'd'], ['a', 'e', 'f', 'g'], ['h', 'i', 'j', 'k'], ['h', 'l', 'm', 'n']]

因此，您的解决方案需要 3 个步骤：

使用pass_row 映射您的数据集
用sum内置函数减少结果，应用于初始累加器[]
将结果提供给 Spark DataFrame

在普通的 Python 中，下面的单行代码就可以完成这项工作

>>> fn = lambda ls : sum([ [ i[0] + el.split(',') for el in i[1]] for i in ls ], [])
>>> fn([[['a'],['b,c,d','e,f,g']],[['h'],['i,j,k','l,m,n']]])
[['a', 'b', 'c', 'd'], ['a', 'e', 'f', 'g'], ['h', 'i', 'j', 'k'], ['h', 'l', 'm', 'n']]

【讨论】：