title: udf for loop in pySpark
body: <p>The piece of code below tries to do the following:</p>
  
  <p>For each <code>customer_code</code> in <code>sdf1</code>, check if this customer code appears in <code>sdf2</code>. If it does, replace the <code>df1.actual_related_customer</code> with the <code>df2.actual_related_customer</code>. </p>
  
  <p>This code is not working because I access my rows in <code>df2</code> wrongly. How can I achieve the above goal? (if you have another suggestion than indices, shoot!) </p>
  
  <pre class="lang-python prettyprint-override"><code>sdf1 = sqlCtx.createDataFrame(
      [
          ('customer1', 'customer_code1', 'other'),
          ('customer2', 'customer_code2', 'other'),
          ('customer3', 'customer_code3', 'other'),
          ('customer4', 'customer_code4', 'other')
      ],
      ('actual_related_customer', 'customer_code', 'other')
  )
  
  sdf2 = sqlCtx.createDataFrame(
      [
          ('Peter', 'customer_code1'),
          ('Deran', 'customer_code5'),
          ('Christopher', 'customer_code3'),
          ('Nick', 'customer_code4')
      ],
      ('actual_related_customer', 'customer_code')
  )
  
  def right_customer(x,y):
      for row in sdf2.collect() :
          if x == row['customer_code'] :
              return row['actual_related_customer']
      return y
  
  fun1 = udf(right_customer, StringType())
  test = sdf1.withColumn(
      "actual_related_customer",
      fun1(sdf1.customer_code, sdf1.actual_related_customer)
  )
  </code></pre>
  
  <p>And my desired output would look like:</p>
  
  <pre class="lang-python prettyprint-override"><code>desired_output = sqlCtx.createDataFrame(
      [
          ('Peter', 'customer_code1', 'other'),
          ('customer2', 'customer_code2', 'other'),
          ('Christopher', 'customer_code3', 'other'),
          ('Nick', 'customer_code4', 'other')
      ],
      ('actual_related_customer', 'customer_code', 'other')
  )
  </code></pre>