UDF Python Function/Operation Assembly () processes data using Python-based functions defined by the user.
It processes one input record at a time and may emit zero or more output records.

Parameters of the User Defined Python Script Function

• Python File
  The file path containing the Python code.
  In a Hadoop YARN environment, this path must be in the Hadoop Distributed File System (HDFS).
• Parameter
  A parameter that will be passed to the Python code
• Output Fields
  List of field names and field types

Sample of UDF Python Function Parameters Dialog in HGrid247-3.5



The Python code specified in the Python File may be written in either procedural or object-oriented programming style.

1. Procedural Programming Style

In procedural programming style, the Python code must define a function named operate.

• The operate function takes one parameter, which is a list of objects representing the input record.
• The function returns:
• a single object,
• a list of objects, or
• a list of lists of objects.
• The operate function is invoked by HGrid247 for each record during the processing phase.

Optionally, the Python code may also define a function named prepare that takes a single string parameter.
The prepare function is invoked by HGrid247 during the startup process.

Example of a user-defined Python code function in procedural programming style :

    from datetime import datetime

    def operate(list) :
            global index, format
            output=[]
            count=0
            while (count<len(list)) :
                    if (count==index):
                            date_input=list[count]
                            datetimeobject = datetime.strptime(date_input,'%Y-%m-%d')
                            new_format = datetimeobject.strftime(format)
                            output.append(new_format)
                    else :
                            output.append(list[count])	  
                    count=count+1

            return output


    def prepare(param) :
            global index, format
            arr_param=param.split(',')
            index=int(arr_param[0])
            format=arr_param[1]
   

2. Object Oriented Programming Style

In object-oriented programming style, the Python code must define a class that inherits from HGridFunction.

• The class must define a constructor that takes a single string parameter.
• The class must override the parent class’s operate method, which takes one parameter:
  • a list of objects representing the input record.
• The operate method returns:
  • a single object,
  • a list of objects, or
  • a list of lists of objects.
• The operate method is invoked by HGrid247 for each record during processing.

Example of a User-Defined Python Code Function (Object-Oriented Programming Style)

    from datetime import datetime
    class DateFunction(HGridFunction):
            def __init__(self,param):
                    arr_param = param.split(',')
                    self.index=int(arr_param[0])
                    self.format=arr_param[1]

            def operate(self,list) :
                    output=[]
                    count=0
                    while (count<len(list)) :
                            if (count==self.index):
                                    date_input=list[count]
                                    datetimeobject = datetime.strptime(date_input,'%Y-%m-%d')
                                    new_format = datetimeobject.strftime(self.format)
                                    output.append(new_format)
                            else :
                                    output.append(list[count])	  
                            count=count+1

                    return output
   

---

Supported Engines

Map Reduce	Spark 1.x RDD	Spark 2.x RDD	Spark 2.x Dataset	Apache Beam 2.29.0 Batch	Apache Beam 2.29.0 Streaming	Apache Flink 1.12.0 Batch	Apache Flink 1.12.0 Streaming	Apache Tez 0.9.1	PySpark

Note: Do not forget to include the Jython library (jython-standalone-2.7.3.jar) when submitting the job.

On mapreduce job :
hadoop jar <project-name>_mr.jar ClassToRun -libjars <JYTHON-PATH>/jython-standalone-2.7.3.jar input output
For example :

        hadoop jar workshop_mr.jar workshop.python.TestPython -libjars /root/jython-standalone-2.7.3.jar
   

On spark job :
spark-submit --class ClassToRun --master yarn --jars <JYTHON-PATH>/jython-standalone-2.7.3.jar    <project-name>_spark2.jar input output
For example :

        spark-submit  --class workshop.python.TestPython --master yarn  --jars /root/jython-standalone-2.7.3.jar workshop_spark2.jar
   

On Flink job :
flink run -c ClassToRun --classpath file:///<JYTHON-PATH>/jython-standalone-2.7.3.jar    -j <project-name>_flink.jar --input_option input_path --output_option output_path
For example :

        flink run -c  workshop.python.TestPython --classpath file:///root/jython-standalone-2.7.3.jar -j workshop_flink.jar --hfs_source_1 file:///D:/hgrid/input/sample_data  --hfs_sink_1 file:///D:/hgrid/test/python_out 
   

On Beam job Direct-Runner on Windows :
java -cp <project-name>_beam.jar;<HGridApp-Path>\HGrid247-3.5\BeamRunner\direct-runner/*;<JYTHON-PATH>\jython-standalone-2.7.3.jar ClassToRun --input_option=input_path --output_option=output_path
For example :

        java  -cp workshop_beam.jar;D:\BigDataTool\HGrid247-3.5\BeamRunner\direct-runner/*;D:\jython\jython-standalone-2.7.3.jar workshop.python.TestPython --hfs_source_1=D:/hgrid/input/sample_data  --hfs_sink_1=D:/hgrid/test/python_out 
   

On Beam job Direct-Runner on Linux :
java -cp <project-name>_beam.jar:<HGridApp-Path>/HGrid247-3.5/BeamRunner/direct-runner/*:<JYTHON-PATH>/jython-standalone-2.7.3.jar ClassToRun --input_option=input_path --output_option=output_path
For example :

        java  -cp workshop_beam.jar:/root/HGrid247-3.5/BeamRunner/direct-runner/*:/root/jython/jython-standalone-2.7.3.jar workshop.python.TestPython --hfs_source_1=/root/hgrid/input/sample_data  --hfs_sink_1=/root/hgrid/test/python_out