@SuppressWarnings("unchecked")
	public static <X> X deserializeFunction(RuntimeContext context, byte[] serFun) throws FlinkException {
		if (!jythonInitialized) {
			// This branch is only tested by end-to-end tests
			String path = context.getDistributedCache().getFile(PythonConstants.FLINK_PYTHON_DC_ID).getAbsolutePath();

			String scriptName = PythonStreamExecutionEnvironment.PythonJobParameters.getScriptName(context.getExecutionConfig().getGlobalJobParameters());

			try {
				initPythonInterpreter(
					new String[]{Paths.get(path, scriptName).toString()},
					path,
					scriptName);
			} catch (Exception e) {

				try {
					LOG.error("Initialization of jython failed.", e);
					throw new FlinkRuntimeException("Initialization of jython failed.", e);
				} catch (Exception ie) {
					// this may occur if the initial exception relies on jython being initialized properly
					LOG.error("Initialization of jython failed. Could not print original stacktrace.", ie);
					throw new FlinkRuntimeException("Initialization of jython failed. Could not print original stacktrace.");
				}
			}
		}

		try {
			return (X) SerializationUtils.deserializeObject(serFun);
		} catch (IOException | ClassNotFoundException ex) {
			throw new FlinkException("Deserialization of user-function failed.", ex);
		}
	}
@ SuppressWarnings ( "unchecked" ) public static < X > X deserializeFunction ( RuntimeContext context , byte [ ] serFun ) throws FlinkException { if ( ! jythonInitialized ) { String path = context . getDistributedCache ( ) . getFile ( PythonConstants . FLINK_PYTHON_DC_ID ) . getAbsolutePath ( ) ; String scriptName = PythonStreamExecutionEnvironment . PythonJobParameters . getScriptName ( context . getExecutionConfig ( ) . getGlobalJobParameters ( ) ) ; try { initPythonInterpreter ( new String [ ] { Paths . get ( path , scriptName ) . toString ( ) } , path , scriptName ) ; } catch ( Exception e ) { try { LOG . error ( "Initialization of jython failed." , e ) ; throw new FlinkRuntimeException ( "Initialization of jython failed." , e ) ; } catch ( Exception ie ) { LOG . error ( "Initialization of jython failed. Could not print original stacktrace." , ie ) ; throw new FlinkRuntimeException ( "Initialization of jython failed. Could not print original stacktrace." ) ; } } } try { return ( X ) SerializationUtils . deserializeObject ( serFun ) ; } catch ( IOException | ClassNotFoundException ex ) { throw new FlinkException ( "Deserialization of user-function failed." , ex ) ; } }
Deserialize the given python function. If the functions class definition cannot be found we assume that this is
the first invocation of this method for a given job and load the python script containing the class definition
via jython.

@param context the RuntimeContext of the java function
@param serFun serialized python UDF
@return deserialized python UDF
@throws FlinkException if the deserialization failed