title: Spark UDF not running in parallel
body: <p>I'm attempting to use the Python port of the Google <code>phonenumbers</code> library to normalize 50 Million phone numbers. I'm reading into a SparkDataFrame from a Parquet file on S3 and then running operations on the dataframe. The following function, <code>parsePhoneNumber</code>, is expressed as a UDF:</p>
  
  <pre><code>def isValidNumber(phoneNum):
      try:
          pn = phonenumbers.parse(phoneNum, "US")
      except:
          return False
      else:
          return phonenumbers.is_valid_number(pn) and phonenumbers.is_possible_number(pn)
  
  def parsePhoneNumber(phoneNum):
      if isValidNumber(phoneNum):
          parsedNumber = phonenumbers.parse(phoneNum, "US")
          formattedNumber = phonenumbers.format_number(parsedNumber, phonenumbers.PhoneNumberFormat.E164)
  
          return (True, parsedNumber.country_code, formattedNumber, parsedNumber.national_number, parsedNumber.extension)
      else:
          return (False, None, None, None)
  </code></pre>
  
  <p>And below is a sample of how I use the UDF to derive new columns:</p>
  
  <pre><code>newDataFrame = oldDataFrame.withColumn("new_column", parsePhoneNumber_udf(oldDataFrame.phone)).select("id", "new_column".national_number)
  </code></pre>
  
  <p>Executing the UDF by running <code>display(newDataFrame)</code> or <code>newDataFrame.show(5)</code> or something similar only uses one executer in the cluster, so it doesn't appear that something in the UDF is causing it only run on one worker. </p>
  
  <p>If I'm doing anything that would prevent this from running in parallel, can you provide some insight?</p>
  
  <p>The execution environment is on a cloud cluster controlled by Databricks.</p>
  
  <p>Edit: Below is the output of <code>oldDataFrame.explain</code></p>
  
  <pre><code>== Parsed Logical Plan ==
  Relation[id#621,person_id#622,phone#623,type#624,source_id#625,created_date#626,modified_date#627] parquet
  
  == Analyzed Logical Plan ==
  id: string, person_id: string, phone: string, type: string, source_id: string, created_date: string, modified_date: string
  Relation[id#621,person_id#622,phone#623,type#624,source_id#625,created_date#626,modified_date#627] parquet
  
  == Optimized Logical Plan ==
  Relation[id#621,person_id#622,phone#623,type#624,source_id#625,created_date#626,modified_date#627] parquet
  
  == Physical Plan ==
  *FileScan parquet [id#621,person_id#622,phone#623,type#624,source_id#625,created_date#626,modified_date#627] Batched: true, Format: Parquet, Location: InMemoryFileIndex[dbfs:/mnt/person-data/parquet/phone], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;id:string,person_id:string,phone:string,type:string,source_id:string,created_date:strin...
  </code></pre>