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.setProcessor(wrapper.processor);
  		this.outputStreams = wrapper.outputStreams;
  		
  		// Init Processor and Streams
  		this.getProcessor().onCreate(0);
  		for (SamzaStream stream:this.outputStreams) {
  			stream.onCreate();
  		}
  		
  	}
docs: /*
  Implement Samza task