title: Dealing with Missing Data in RDD using PySpark
body: <p>I am having some issues cleaning up with my data, particularly with missing data.</p>
  
  <p>After parsing my data file and doing preliminary data cleaning, a snapshot of the data is as follows:</p>
  
  <pre><code>r2.take(5)
  Out[210]: 
  [u'2016-10-26 13.43.06 C03 R27 80327 ONYX E 2605',
   u'2016-10-26 18.16.57 C04 R34 80544 GOLD E 7094',
   u'2016-10-26 18.24.15 C04 R33 80544 GOLD E 6875',
   u'2016-10-26 17.43.57 C03 R19 80908 ONYX E 5214',
   u'2016-10-26 17.07.55 C03 R27 80436 GOLD E 3436']
  </code></pre>
  
  <p>Based off Spark tutorials, I do the following next steps to [eventually] create a Spark Data frame (I omitted some steps):</p>
  
  <pre><code># Create RDD
  from pyspark.sql import SQLContext, Row
  
  # Load text files
  parts = r2.map(lambda l: l.split(" "))
  apachelogs = parts.map(lambda p: Row(
    date=p[0],
    time=p[1],
    category_code=p[2],
    item_code=p[3],
    purchase_ref=p[4],
    selling_team=p[5],
    language=p[6],
    gross_sales=p[7]))
  </code></pre>
  
  <p>After a few more steps, I attempt to run simple commands like .groupBy().count().collect() and run into errors. When I went back into my data, I noticed I had missing data, particularly, when I run the following:</p>
  
  <pre><code>parts.filter(lambda x: len(x) != 8).take(12)
  
  Out[236]: 
  [[u'2016-10-26', u'10.25.26', u'C04', u'800779', u'BLUE', u'E'],
   [u'2016-10-26', u'11.19.42', u'C03', u'803817', u'ONYX', u'E'],
   [u'2016-10-26', u'11.22.12', u'C04', u'8132705', u'GOLD', u'E'],
   [u'2016-10-26', u'11.22.22', u'C04', u'8154705', u'GOLD', u'E'],
   [u'2016-10-26', u'12.34.22', u'C05', u'8125781', u'ONYX', u'E'],
   [u'2016-10-26', u'12.40.05', u'C05', u'812381', u'ONYX', u'E'],
   [u'2016-10-26', u'22.16.42', u'C03', u'832347', u'ONYX', u'E'],
   [u'2016-10-26', u'18.07.36', u'C01', u'852300', u'GOLD', u'E'],
   [u'2016-10-26', u'11.26.39', u'C02', u'845424', u'BLUE', u'F'],
   [u'2016-10-26',
    u'12.16.00',
    u'C05',
    u'R39',
    u'845436668',
    u'BLUE',
    u'6975'],
   [u'2016-10-26', u'11.26.58', u'C05', u'83344280', u'GOLD', u'F']]
  </code></pre>
  
  <p>As you can see we're missing observations for the following columns: item_code, language, and gross_sales.</p>
  
  <p>What I'm trying to do is if else statements (or something similar) so I could fix these columns. I've done this code in a SQL environment but not sure how I would translate this to a PySpark environment.</p>
  
  <p>For example, here is my code in SQL if we were working in a SQL environment:</p>
  
  <pre><code>select date, time, category_code, 
  if(item_code like '8%','',item_code) as item_code,
  
  /* Lets look at this code in the example below*/
  if(item_code like '8%',item_code,if(purchase_ref not in ('GOLD','ONYX'),purchase_ref,'')) as purchase_ref,
  
  if(purchase_ref in ('GOLD','ONYX'),purchase_ref,if(selling_team in ('GOLD','ONYX'),selling_team,'')) as selling_team,
  
  if(selling_team in ('E','F'),selling_team,if(language in ('E','F'),language,'')) as language,
  
  if(gross_sales is null,language,gross_sales) as gross_sales
  </code></pre>
  
  <p>Lets look at one example. <code>purchase_ref</code> always begins with a number "8". For the <code>purchase_ref</code> column, if the <code>item_code</code> contains the <code>purchase_ref</code> number due to the values being shifted over because of missing data, then its result should be that number. If it contains a <code>selling_team</code> value, leave it blank.</p>
  
  <p>The SQL code worked when I was attempting to fix the rows that had missing data. </p>
  
  <p>My question is: 
  How would I approach cleaning my data from this issue? </p>
  
  <p>Also, would it be more feasible if I did it in a Spark Data Frame instead (I can imagine how I would approach this using Pandas <code>np.where</code> but not sure how that translates here)?</p>
  
  <p>Thanks.</p>
  
  <p><strong>EDIT:</strong></p>
  
  <p>To be more clear on my preliminary steps, they are the following:</p>
  
  <p>Import text files from AWS. Here is what it looks like raw:</p>
  
  <pre><code>myApacheLogs.take(25)
  Out[227]: 
  [u'  2016-10-26  13.43.06  C03    R27       84327  ONYX  E     2605                                        ',
   u'',
   u'  2016-10-26  18.16.57  C04    R34       896544  GOLD  E     7094                                        ',
   u'',
   u'  2016-10-26  18.24.15  C04    R33       867544  GOLD  E     6875                                        ',
   u'',
   u'  2016-10-26  17.43.57  C03    R19       8776908  ONYX  E     5214                                        ',
   u'',
   u'  2016-10-26  17.07.55  C03    R27       8654436  GOLD  E     3436                                        ']
  </code></pre>
  
  <p>I did some cleaning with the following:</p>
  
  <pre><code>r1 = myApacheLogs.filter(lambda x: x is not u'').filter(lambda x: x is not u'')
  r1.take(10)
  Out[209]: 
  [u'  2016-10-26  13.43.06  C03    R27       84327  ONYX  E     2605                                        ',
   u'  2016-10-26  18.16.57  C04    R34       84544  GOLD  E     7094                                        ',
   u'  2016-10-26  18.24.15  C04    R33       84544  GOLD  E     6875                                        ',
   u'  2016-10-26  17.43.57  C03    R19       895408  ONYX  E     5214                                        ',
   u'  2016-10-26  17.07.55  C03    R27       86436  GOLD  E     3436                                        ',
   u'  2016-10-26  13.01.30  C05    R43       248945  GOLD  E     6956                                        ',
   u'  2016-10-26  10.07.10  C03    R29       863356  GOLD  E     0861                                        ',
   u'  2016-10-26  11.31.19  C04    R32       865684  ONYX  E     2595                                        ',
   u'  2016-10-26  13.31.33  C01    RR1       865690  ONYX  F     7093                                        ',
   u'  2016-10-26  14.06.37  C03    R20       865243  GOLD  E     6959                                        ']
  
  def myFunc(s):
    results = re.sub("\s\s+" , " ", s) # Properly space columns
    results = results.lstrip() # Remove whitespace on the left
    results = results.rstrip() # Remove whitespace on the right
    return results
  r2 = r1.map(myFunc)
  </code></pre>