@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();
		}
		
	}
@ 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 ) 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 ; this . getProcessor ( ) . onCreate ( 0 ) ; for ( SamzaStream stream : this . outputStreams ) { stream . onCreate ( ) ; } }
/*
Implement Samza task