Nsryjdtyk

I have RDD in this form:

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

What I've wanted to achieve:

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

What I did:

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)

It works for smaller samples of data. However, I am getting list index out of range on line for el in line[1]: when I try to run it on my whole dataset...

Basically I have one key (lets say ['a']) for ~100 different sets of values in ['b,c,d','e,f,g']. My ultimate goal is to have it as spark dataframe in form of rows:

col1 col2 col3 col 4

a     b    c    d

a     e    f    g

h     i    j    k

h     l    m    n

Thank you for any advice!

Your error seems more related to your data than your function (which seems to be correct, although a bit overcomplicated), and it looks like you applied it to a line that didn't have a line[1].

Could you make sure that the number of elements of line is constant in your actual dataset, for example with :

def pass_row(line):

    assert len(line) == 2

    return [ "%s, %s" % (''.join(line[0]), el) for el in line[1]]

That being said, for your actual goal, you probably should stop dealing with strings from that point and directly get your data as a 2D-array, for example with :

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']]]

Warning here, you can't directly feed a DataFrame with that solution, as each group of prefix-generated lines is still nested in its own list (it's only a "list of 2D-arrays").

For example, using sum function, that you could easily transpose in a reduce step :

>>> sum(b, )

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

Your solution would thus need 3 steps :

• map your dataset with pass_row as you do


• reduce the result with sum built-in function, applied to the initial accumulator
  


• feed the result to a Spark DataFrame

In plain Python, the following one-liner would do the job

>>> 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']]