code: @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);
  		}
  	}
docs: 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