examples/mroperator/src/main/java/org/apache/apex/examples/mroperator/LogCountsPerHour.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.apex.examples.mroperator;

import java.io.IOException;
import java.util.Calendar;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * <p>LogCountsPerHour class.</p>
 *
 * @since 0.9.0
 */
public class LogCountsPerHour extends Configured implements Tool
{

  public static class LogMapClass extends MapReduceBase
      implements Mapper<LongWritable, Text, DateWritable, IntWritable>
  {
    private DateWritable date = new DateWritable();
    private static final IntWritable one = new IntWritable(1);

    public void map(LongWritable key, Text value, OutputCollector<DateWritable, IntWritable> output, Reporter reporter) throws IOException
    {
      // Get the value as a String; it is of the format:
      // 111.111.111.111 - - [16/Dec/2012:05:32:50 -0500] "GET / HTTP/1.1" 200 14791 "-" "Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)"
      String text = value.toString();

      // Get the date and time
      int openBracket = text.indexOf('[');
      int closeBracket = text.indexOf(']');
      if (openBracket != -1 && closeBracket != -1) {
        // Read the date
        String dateString = text.substring(text.indexOf('[') + 1, text.indexOf(']'));

        // Build a date object from a string of the form: 16/Dec/2012:05:32:50 -0500
        int index = 0;
        int nextIndex = dateString.indexOf('/');
        int day = Integer.parseInt(dateString.substring(index, nextIndex));

        index = nextIndex;
        nextIndex = dateString.indexOf('/', index + 1);
        String month = dateString.substring(index + 1, nextIndex);

        index = nextIndex;
        nextIndex = dateString.indexOf(':', index);
        int year = Integer.parseInt(dateString.substring(index + 1, nextIndex));

        index = nextIndex;
        nextIndex = dateString.indexOf(':', index + 1);
        int hour = Integer.parseInt(dateString.substring(index + 1, nextIndex));

        // Build a calendar object for this date
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.DATE, day);
        calendar.set(Calendar.YEAR, year);
        calendar.set(Calendar.HOUR, hour);
        calendar.set(Calendar.MINUTE, 0);
        calendar.set(Calendar.SECOND, 0);
        calendar.set(Calendar.MILLISECOND, 0);

        if (month.equalsIgnoreCase("dec")) {
          calendar.set(Calendar.MONTH, Calendar.DECEMBER);
        } else if (month.equalsIgnoreCase("nov")) {
          calendar.set(Calendar.MONTH, Calendar.NOVEMBER);
        } else if (month.equalsIgnoreCase("oct")) {
          calendar.set(Calendar.MONTH, Calendar.OCTOBER);
        } else if (month.equalsIgnoreCase("sep")) {
          calendar.set(Calendar.MONTH, Calendar.SEPTEMBER);
        } else if (month.equalsIgnoreCase("aug")) {
          calendar.set(Calendar.MONTH, Calendar.AUGUST);
        } else if (month.equalsIgnoreCase("jul")) {
          calendar.set(Calendar.MONTH, Calendar.JULY);
        } else if (month.equalsIgnoreCase("jun")) {
          calendar.set(Calendar.MONTH, Calendar.JUNE);
        } else if (month.equalsIgnoreCase("may")) {
          calendar.set(Calendar.MONTH, Calendar.MAY);
        } else if (month.equalsIgnoreCase("apr")) {
          calendar.set(Calendar.MONTH, Calendar.APRIL);
        } else if (month.equalsIgnoreCase("mar")) {
          calendar.set(Calendar.MONTH, Calendar.MARCH);
        } else if (month.equalsIgnoreCase("feb")) {
          calendar.set(Calendar.MONTH, Calendar.FEBRUARY);
        } else if (month.equalsIgnoreCase("jan")) {
          calendar.set(Calendar.MONTH, Calendar.JANUARY);
        }


        // Output the date as the key and 1 as the value
        date.setDate(calendar.getTime());
        output.collect(date, one);
      }
    }
  }

  public static class LogReduce extends MapReduceBase
      implements Reducer<DateWritable, IntWritable, DateWritable, IntWritable>
  {
    public void reduce(DateWritable key, Iterator<IntWritable> values, OutputCollector<DateWritable, IntWritable> output, Reporter reporter) throws IOException
    {
      // Iterate over all of the values (counts of occurrences of this word)
      int count = 0;
      while (values.hasNext()) {
        // Add the value to our count
        count += values.next().get();
      }

      // Output the word with its count (wrapped in an IntWritable)
      output.collect(key, new IntWritable(count));
    }
  }


  public int run(String[] args) throws Exception
  {
    // Create a configuration
    Configuration conf = getConf();

    // Create a job from the default configuration that will use the WordCount class
    JobConf job = new JobConf(conf, LogCountsPerHour.class);

    // Define our input path as the first command line argument and our output path as the second
    Path in = new Path(args[0]);
    Path out = new Path(args[1]);

    // Create File Input/Output formats for these paths (in the job)
    FileInputFormat.setInputPaths(job, in);
    FileOutputFormat.setOutputPath(job, out);

    // Configure the job: name, mapper, reducer, and combiner
    job.setJobName("LogAveragePerHour");
    job.setMapperClass(LogMapClass.class);
    job.setReducerClass(LogReduce.class);
    job.setCombinerClass(LogReduce.class);

    // Configure the output
    job.setOutputFormat(TextOutputFormat.class);
    job.setOutputKeyClass(DateWritable.class);
    job.setOutputValueClass(IntWritable.class);

    // Run the job
    JobClient.runJob(job);
    return 0;
  }

  public static void main(String[] args) throws Exception
  {
    // Start the LogCountsPerHour MapReduce application
    int res = ToolRunner.run(new Configuration(), new LogCountsPerHour(), args);
    System.exit(res);
  }
}