code: @Override
  	public void init(Config config, TaskContext context) throws Exception {
  		String yarnConfHome = config.get(SamzaConfigFactory.YARN_CONF_HOME_KEY);
  		if (yarnConfHome != null && yarnConfHome.length() > 0) // if the property is set , otherwise, assume we are running in
              												// local mode and ignore this
  			SystemsUtils.setHadoopConfigHome(yarnConfHome);
  		
  		String filename = config.get(SamzaConfigFactory.FILE_KEY);
  		String filesystem = config.get(SamzaConfigFactory.FILESYSTEM_KEY);
  		
  		this.setName(config.get(SamzaConfigFactory.JOB_NAME_KEY));
  		SerializationProxy wrapper = (SerializationProxy) SystemsUtils.deserializeObjectFromFileAndKey(filesystem, filename, this.getName());
  		this.setOutputStream(wrapper.outputStream);
  		SamzaStream output = (SamzaStream)this.getOutputStream();
  		if (output != null) // if output stream exists, set it up
  			output.onCreate();
  	}
docs: /*
  Implement Samza Task