Custom Buffer Transformation

UDF Python Buffer Transformation is used to call Python code for Buffer-Transformation.
The parameter of UDFPythonBufferTransformation is Python File , File path contains python-code.
In hadoop yarn environment, this path must be in Hadoop Distributed File System (HDFS).

Sample of UDFPythonBufferTransformation Parameter's Dialog on HGrid247-3.5

Python-code contained in 'Python File' may be in procedural programming style or in object oriented programming style.

1. In Procedural Programming Style

In procedural programming style, python-code must have a function named prepare.
The prepare function has no parameter, it is invoked when starting processing of group records.

The Python-code must has a function named receive, this function has two parameters,the first parameter is a list of objects, and the second parameter is an integer represents the index of selected element of the first parameter. The receive function is invoked when a record is received.

The Python-code must also has a function name emitValue with no parameter. It is invoked when an output will be emitted.

Python-code may have a function named onChangeSubGroup which has two parameters,the first parameter is a list of objects, and the second parameter is an integer represents the index of selected element of the first parameter. The onChangeSubGroup function is invoked when the sub-group value changed.

Python-code may also have a function named afterEmit which has one parameter, a list of objects. The afterEmit function is invoked after a value emitted.

Python-code may also have a function named cleanup with no parameter. This function is invoked when a group of records completely processed.

Example of python-code buffer-transformation-function in procedural programming style :

    def prepare():
		global output
		output=0
    def cleanup():
		global output
		output=0
		
    def receive(list, index):
		global output
		output=output+(list[index]*list[index+1])
		
    def emitValue():
		global output
		return output

2. In Object Oriented Programming Style

In object oriented programming style, python-code must define a class inherited from the class HGridBufferTransform.
This class must define a constructor that takes no parameter.

The class must override its parent's method named receive, this method has two parameters,the first parameter is a list of objects, and the second parameter is an integer represents the index of selected element of the first parameter. The receive method is invoked when a record is received.

The class must also override its parent's method named emitValue with no parameter. It is invoked when an output will be emitted.

The class may override its parent's method named onChangeSubGroup which has two parameters,the first parameter is a list of objects, and the second parameter is an integer represents the index of selected element of the first parameter. The onChangeSubGroup method is invoked when the sub-group value changed.

The class may also override its parent's method named afterEmit which has one parameter, a list of objects. The afterEmit method is invoked after a value emitted.

Example of python-code buffer-transformation-function in object oriented programming style :

    class Accumlation(HGridBufferTransform):
	def __init__(self):
		self.output=0
		
	def onChangeSubGroup(self, list, index):
		pass
		
	def cleanup(self):
		self.output=0
		
	def afterEmit(self,list):
		pass
		
	def receive(self, list, index):
		self.output=self.output+(list[index]*list[index+1])
		
	def emitValue(self):
		return self.output

Note : Don't forget to include jython library (jython-standalone-2.7.3.jar) when submitting 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