UDF Python Buffer Assembly () processes grouped and sorted data/records using a Python-based buffer defined by the user.
It processes a group of records at a time and may emit zero or more output records.

Parameters of the User Defined Python Code Buffer

• 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
• New Output Fields
  A list of new output field names and their corresponding field types.

Sample of UDF Python Buffer Parameter Dialog on 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 lists of objects representing the input group of records.
• 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 group of records during processing.

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 Buffer (Procedural Programming Style).

    def operate(lists) :
            min =None
            max = None
            sum=0
            for list in lists:
                    data =list[4]*list[5]
                    if (min==None):
                            min=data
                    else :
                            if (min>data) :
                                    min=data
                    if (max==None):
                            max=data
                    else :
                            if (max<data):
                                    max=data

                    sum=sum+data

            return [min,max,(sum/len(lists))]
   

2. Object Oriented Programming Style

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

• 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 lists of objects representing the input group of records.
• 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 group of records during processing.

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

    class MinMaxAvgBuffer(HGridBuffer):
            def __init__(self,parameter):
                    self.parameter=parameter	

            def operate(self,lists) :
                    min =None
                    max = None
                    sum=0
                    for list in lists:
                            data =list[4]*list[5]
                            if (min==None):
                                    min=data
                            else :
                                    if (min>data) :
                                            min=data
                            if (max==None):
                                    max=data
                            else :
                                    if (max<data):
                                            max=data

                            sum=sum+data

                    return [min,max,(sum/len(lists))]
   

---

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