Hadoop Netflix数据统计分析2(转)

map阶段
package com.taobao;
import java.io.*;
import java.util.*;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.StringUtils;

public class MyMapper {   
      public static class MapClass extends MapReduceBase
        implements Mapper<LongWritable, Text, Text, Text> {
         
          Path[] localFiles = new Path[0];
          HashMap<String, String> dateMap = new HashMap<String, String>();
           
          public void configure(JobConf job) {
               
                    if(job.getBoolean("HadoopDriver.distributedCacheFile", false)) {
                        try {
                            localFiles = DistributedCache.getLocalCacheFiles(job);
                        }
                        catch (IOException ioe) {
                            System.err.println("Caught exception while getting cached files " + StringUtils.stringifyException(ioe));
                        }
                        if(localFiles[0].toString() != null) {
                            try {