UDF Python с несколькими аргументами - PullRequest
Новая
       13

UDF Python с несколькими аргументами

0 голосов
/ 19 января 2019

Задача состояла в том, чтобы вывести всех руководителей сотрудников на уровень выше в Spark.

Например: в приведенном ниже наборе данных .. 

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

Учитывая, что для работника '1' ожидаемый результат равен ['3', '4', '5', '6', '7', '8', '9']

Ниже приведен код PySpark, который я пытался написать, но который не удался с упомянутой ошибкой. 

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)

Ошибка, как показано ниже: 

df1.select('EMPLOYEE_ID',spark_recur_man('EMPLOYEE_ID',3,list1).alias('heirarchy')).show(truncate=False)

Файл "C: \ opt \ spark \ spark-2.2.0-bin-hadoop2.7 \ python \ lib \ pyspark.zip \ pyspark \ sql \ functions.py", строка 1957 г., в файле оболочки "C: \ opt \ spark \ spark-2.2.0-bin-hadoop2.7 \ python \ lib \ pyspark.zip \ pyspark \ sql \ functions.py", строка 1918, в вызов файл "C: \ неавтоматического \ искра \ искровым 2.2.0-бен-hadoop2.7 \ питон \ Lib \ pyspark.zip \ pyspark \ SQL \ column.py", строка 60 в файле _to_seq "C: \ неавтоматического \ искра \ искровым 2.2.0-бен-hadoop2.7 \ питон \ Lib \ pyspark.zip \ pyspark \ SQL \ column.py", строка 48, в файле _to_java_column "C: \ неавтоматического \ искра \ искровым 2.2.0-бен-hadoop2.7 \ питон \ Lib \ pyspark.zip \ pyspark \ SQL \ column.py", строка 41 в файле _create_column_from_name "C: \ неавтоматического \ искра \ искровым 2.2.0-бен-hadoop2.7 \ питон \ Lib \ py4j-0.10.4-src.zip \ py4j \ java_gateway.py", строка 1133, вызов Файл "C: \ неавтоматического \ искра \ искровым 2.2.0-бен-hadoop2.7 \ питон \ Lib \ pyspark.zip \ pyspark \ SQL \ utils.py", строка 63, в файле deco "C: \ opt \ spark \ spark-2.2.0-bin-hadoop2.7 \ python \ lib \ py4j-0.10.4-src.zip \ py4j \ protocol.py", строка 323, в get_return_value py4j.protocol. Py4JError: ошибка произошло при вызове z: org.apache.spark.sql.functions.col. Трассировка: py4j.Py4JException: метод col ([класс java.lang.Integer]) не делает есть

Может кто-нибудь помочь мне с вышеуказанным вопросом.

Заранее спасибо

1 Ответ

0 голосов
/ 20 января 2019

Spark не может получить тип последнего параметра массива для UDF, то есть list1.

Ниже вы можете попробовать обойти: 

import sys    
import os    
from operator import add    
import re    
import numpy as np
import pyspark.sql.functions as F   
import pyspark.sql.types as t
import pyspark.sql.column as c 

def square(x):
    return x**2

def recur_man(emp_id,lvl,list1):
    if(type(list1)==int):
        list1=[]
    with open("D:\MOCK_DATA\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 *        
    sc = SparkContext.getOrCreate()
    sqlContext = SQLContext(sc)    
    rdd = sc.textFile("D:\MOCK_DATA\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,y,z: recur_man(x,y,z), t.ArrayType(t.StringType()))    
    list1 = lit(0)    
    df1.select('EMPLOYEE_ID',spark_recur_man('EMPLOYEE_ID',lit(3),list1).alias('heirarchy')).show(truncate=False)    
    sc.stop()    
except ImportError as e:    
    print ("Error importing Spark Modules", e)    
    sys.exit(1)
...