public synchronized static void setJobSpecificOutputPaths ( State state ) { // Other tasks may have set this already if ( ! StringUtils . containsIgnoreCase ( state . getProp ( ConfigurationKeys . WRITER_STAGING_DIR ) , state . getProp ( ConfigurationKeys . JOB_ID_KEY ) ) ) { state . setProp ( ConfigurationKeys . WRITER_STAGING_DIR , new Path ( state . getProp ( ConfigurationKeys . WRITER_STAGING_DIR ) , state . getProp ( ConfigurationKeys . JOB_ID_KEY ) ) ) ; state . setProp ( ConfigurationKeys . WRITER_OUTPUT_DIR , new Path ( state . getProp ( ConfigurationKeys . WRITER_OUTPUT_DIR ) , state . getProp ( ConfigurationKeys . JOB_ID_KEY ) ) ) ; } }
Each job gets its own task - staging and task - output directory . Update the staging and output directories to contain job_id . This is to make sure uncleaned data from previous execution does not corrupt final published data produced by this execution .