title: Python UDF with multiple arguments
body: <p>Problem statement was to get all managers of employees upto a given level in Spark.</p>
  
  <p>For example: in the below dataset..</p>
  
  <pre><code>EMPLOYEE_ID,FIRST_NAME,LAST_NAME,EMAIL,PHONE_NUMBER,HIRE_DATE,JOB_ID,SALARY,COMMISSION_PCT,MANAGER_ID,DEPARTMENT_ID
  1,Donald,OConnell,DOCONNEL,650.507.9833,21/06/2007,SH_CLERK,2600,,2,500
  2,Douglas,Grant,DGRANT,650.507.9844,13/01/2008,SH_CLERK,2600,,3,50
  3,Jennifer,Whalen,JWHALEN,515.123.4444,17/09/2003,AD_ASST,4400,,4,10
  4,Michael,Hartstein,MHARTSTE,515.123.5555,17/02/2004,MK_MAN,13000,,5,20
  5,Pat,Fay,PFAY,603.123.6666,17/08/2005,MK_REP,6000,,6,20
  6,Susan,Mavris,SMAVRIS,515.123.7777,07/06/2002,HR_REP,6500,,7,40
  7,Hermann,Baer,HBAER,515.123.8888,07/06/2002,PR_REP,10000,,8,70
  8,Shelley,Higgins,SHIGGINS,515.123.8080,07/06/2002,AC_MGR,12008,,9,110
  9,William,Gietz,WGIETZ,515.123.8181,07/06/2002,AC_ACCOUNT,8300,,,110
  </code></pre>
  
  <p>Given employee '1' result expected is ['3', '4', '5', '6', '7', '8', '9']</p>
  
  <p>Below is the PySpark code which I tried writing which however failed with the mentioned error.</p>
  
  <pre><code>import sys    
  import os    
  from operator import add    
  import re    
  import pyspark.sql.functions as F    
  
  os.environ['SPARK_HOME'] = "path"    
  sys.path.append("path")    
  sys.path.append("path")    
  
  def recur_man(emp_id,lvl,list1):    
      with open("path\\employee_1.txt") as f:    
              for lines in f:    
                  if lines.split(',')[0] == emp_id:    
                      list1.append(lines.split(',')[9])    
                      lvl-=1    
                      recur_man(lines.split(',')[9],lvl,list1)    
      return list1    
  
  
  try:    
      from pyspark import SparkContext    
      from pyspark import SparkConf    
      from pyspark.sql import SQLContext    
      from pyspark.sql.functions import *    
      from pyspark.sql.types import *    
      config = SparkConf().setAll([('spark.num.executors','10'),('spark.ui.port','4050')])    
      sc = SparkContext(conf=config)    
      sqlContext = SQLContext(sc)    
      rdd = sc.textFile("path\\employee_1.txt")    
      header = rdd.first()     
      header_mod = [x.encode("utf-8") for x in header.split(',')]    
      rdd = rdd.filter(lambda line: line!=header)    
      rdd = rdd.map(lambda line: line.split(','))    
      df1 = rdd.toDF(header_mod)    
      spark_recur_man = udf(lambda x: recur_man,ArrayType(StringType()))    
      list1 = []    
      df1.select('EMPLOYEE_ID',spark_recur_man('EMPLOYEE_ID',3,list1).alias('heirarchy')).show(truncate=False)    
      sc.stop()    
  except ImportError as e:    
      print ("Error importing Spark Modules", e)    
      sys.exit(1)    
  </code></pre>
  
  <p>Error as below:</p>
  
  <pre><code>df1.select('EMPLOYEE_ID',spark_recur_man('EMPLOYEE_ID',3,list1).alias('heirarchy')).show(truncate=False)
  </code></pre>
  
  <blockquote>
    <p>File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\functions.py", line 1957, in wrapper   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\functions.py", line 1918, in <strong>call</strong>   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\column.py",
    line 60, in _to_seq   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\column.py",
    line 48, in _to_java_column   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\column.py",
    line 41, in _create_column_from_name   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\py4j-0.10.4-src.zip\py4j\java_gateway.py",
    line 1133, in <strong>call</strong>   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\sql\utils.py",
    line 63, in deco   File
    "C:\opt\spark\spark-2.2.0-bin-hadoop2.7\python\lib\py4j-0.10.4-src.zip\py4j\protocol.py", line 323, in get_return_value py4j.protocol.Py4JError: An error
    occurred while calling z:org.apache.spark.sql.functions.col. Trace:
    py4j.Py4JException: Method col([class java.lang.Integer]) does not
    exist</p>
  </blockquote>
  
  <p>Can anyone help me with the above issue.</p>
  
  <p>Thanks in advance</p>