src/main/java/org/lwes/journaller/SequenceDejournaller.java

   1 package org.lwes.journaller;
   2 /**
   3  * @author fmaritato
   4  */
   5
   6 import org.apache.commons.logging.Log;
   7 import org.apache.commons.logging.LogFactory;
   8 import org.apache.hadoop.conf.Configuration;
   9 import org.apache.hadoop.fs.FileSystem;
  10 import org.apache.hadoop.fs.Path;
  11 import org.apache.hadoop.io.BytesWritable;
  12 import org.apache.hadoop.io.NullWritable;
  13 import org.apache.hadoop.io.SequenceFile;
  14 import org.kohsuke.args4j.CmdLineException;
  15 import org.lwes.Event;
  16 import org.lwes.EventSystemException;
  17 import org.lwes.db.EventTemplateDB;
  18
  19 import java.io.IOException;
  20
  21 public class SequenceDejournaller extends DeJournaller {
  22
  23     private static transient Log log = LogFactory.getLog(SequenceDejournaller.class);
  24
  25     public void run() {
  26         Configuration conf = new Configuration();
  27         FileSystem fs = null;
  28         SequenceFile.Reader reader = null;
  29         try {
  30             if (log.isDebugEnabled()) {
  31                 log.debug("Opening file: " + getFileName());
  32             }
  33             fs = FileSystem.get(conf);
  34             Path p = new Path(getFileName());
  35             reader = new SequenceFile.Reader(fs, p, conf);
  36
  37             BytesWritable key = (BytesWritable) reader.getKeyClass().newInstance();
  38             NullWritable value = NullWritable.get();
  39
  40             EventTemplateDB templ = new EventTemplateDB();
  41             while (reader.next(key, value)) {
  42                 Event evt = new Event(key.getBytes(), false, templ);
  43                 if (log.isDebugEnabled()) {
  44                     log.debug("read a k/v: "+evt.toOneLineString());
  45                 }
  46                 handleEvent(evt);
  47             }
  48         }
  49         catch (IOException e) {
  50             log.error(e.getMessage(), e);
  51         }
  52         catch (InstantiationException e) {
  53             log.error(e.getMessage(), e);
  54         }
  55         catch (IllegalAccessException e) {
  56             log.error(e.getMessage(), e);
  57         }
  58         catch (EventSystemException e) {
  59             log.error(e.getMessage(), e);
  60         }
  61         finally {
  62             if (reader != null) {
  63                 try {
  64                     reader.close();
  65                 }
  66                 catch (IOException e) {
  67                     log.error(e.getMessage(), e);
  68                 }
  69             }
  70         }
  71     }
  72
  73     public static void main(String[] args) {
  74         SequenceDejournaller dj = new SequenceDejournaller();
  75         try {
  76             dj.parseArguments(args);
  77         }
  78         catch (CmdLineException e) {
  79             log.error(e.getMessage(), e);
  80         }
  81         dj.run();
  82     }
  83 }