title: PYSPARK UDF to explode records based on date range
body: <p>I am a Noob in Python &amp; Pyspark.
  I need to explode a row of patient into yearly dates, such that each patient has 1 row per year.</p>
  <p>I wrote a python function (below), and registered it as pyspark UDF (having read many articles here). My problem is that when I apply it on my pyspark dataframe, it fails. My function returns 4 values, but I could not figure out how to use/apply using .withColumn or other method/function using this UDF.</p>
  <p>While registering the python function as UDF, I could not understand what data type to be returned. I am sure i messed up somewhere, but dont know where.</p>
  <p>Please help</p>
  <pre><code>import sys
  import pyspark.sql.functions as f
  import pyspark.sql as t
  from pyspark.sql.types import *
  import datetime
  import dateutil
  
  def process_data(identifier, eff_date, end_date):
      eff_year = int(eff_date.split(&quot;/&quot;)[2])
      end_year = int(end_date.split(&quot;/&quot;)[2])
      
      current_year = eff_year
      while current_year &lt;= end_year:
          
          if str(current_year) in eff_date:
              first_date_in_year = eff_date
          else:
              first_date_in_year = &quot;%02d/%02d/%4d&quot; % (1, 1, current_year)
              
          if str(current_year) in end_date:
              last_date_in_year = end_date
          else:
              last_date_in_year = &quot;%02d/%02d/%4d&quot; % (12, 31, current_year)
              
          print (&quot;|&quot;.join([identifier, first_date_in_year, last_date_in_year]))
          current_year += 1
  
          return identifier, first_date_in_year, last_date_of_year, current_year
  
  
  def main():
      
      with open('date.input') as f:
          for line in f.read().splitlines():
              identifier, eff_date, end_date = line.split('|')
              process_data(identifier, eff_date, end_date)
  
  process_data_UDF=f.udf(process_data, StringType())
  
  df_explode=df.withColumn(&quot;NEWCOLS&quot;, process_data_UDF(&quot;INDV_ID&quot;, &quot;EFF_DATE&quot;, &quot;END_DATE&quot;))
  df_explode.show()
  </code></pre>
  <p><strong>Input Data</strong></p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>Identifier</th>
  <th>Eff_Date</th>
  <th>End_Date</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>A0001</td>
  <td>2/1/2019</td>
  <td>3/31/2021</td>
  </tr>
  <tr>
  <td>B0001</td>
  <td>6/1/2020</td>
  <td>11/30/2020</td>
  </tr>
  <tr>
  <td>C0001</td>
  <td>6/1/2020</td>
  <td>1/31/2021</td>
  </tr>
  <tr>
  <td>D0001</td>
  <td>6/1/2020</td>
  <td>12/31/2021</td>
  </tr>
  <tr>
  <td>E0001</td>
  <td>1/1/2019</td>
  <td>6/30/2019</td>
  </tr>
  <tr>
  <td>E0001</td>
  <td>1/1/2020</td>
  <td>12/31/2020</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p><strong>Output Data</strong></p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>Identifier</th>
  <th>New_Eff_Date</th>
  <th>New_End_Date</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>A0001</td>
  <td>2/1/2019</td>
  <td>12/31/2019</td>
  </tr>
  <tr>
  <td>A0001</td>
  <td>01/01/2020</td>
  <td>12/31/2020</td>
  </tr>
  <tr>
  <td>A0001</td>
  <td>01/01/2021</td>
  <td>3/31/2021</td>
  </tr>
  <tr>
  <td>B0001</td>
  <td>6/1/2020</td>
  <td>11/30/2020</td>
  </tr>
  <tr>
  <td>C0001</td>
  <td>6/1/2020</td>
  <td>12/31/2020</td>
  </tr>
  <tr>
  <td>C0001</td>
  <td>01/01/2021</td>
  <td>1/31/2021</td>
  </tr>
  <tr>
  <td>D0001</td>
  <td>6/1/2020</td>
  <td>12/31/2020</td>
  </tr>
  <tr>
  <td>D0001</td>
  <td>01/01/2021</td>
  <td>12/31/2021</td>
  </tr>
  <tr>
  <td>E0001</td>
  <td>1/1/2019</td>
  <td>6/30/2019</td>
  </tr>
  <tr>
  <td>E0001</td>
  <td>1/1/2020</td>
  <td>12/31/2020</td>
  </tr>
  </tbody>
  </table>
  </div>