// Path: src/main/java/com/knewton/mapreduce/SSTableCellMapper.java
// public abstract class SSTableCellMapper<K1, V1,
//                                         K2 extends WritableComparable<?>, V2 extends Writable>
//         extends AbstractSSTableMapper<Cell, K1, V1, K2, V2> {
// 
//     public SSTableCellMapper() {
//         super(Cell.class);
//     }
// 
// }
// 
// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }
// 
// Path: src/main/java/com/knewton/mapreduce/util/CounterConstants.java
// public class CounterConstants {
//     public static final String STUDENT_EVENTS_JOB = "STUDENT_EVENTS";
//     public static final String STUDENT_EVENTS_SKIPPED = "STUDENT_EVENTS_SKIPPED";
//     public static final String STUDENT_EVENTS_COUNT = "STUDENT_EVENTS_COUNT";
//     public static final String STUDENT_EVENTS_PROCESSED = "STUDENT_EVENTS_PROCESSED";
//     public static final String STUDENT_EVENT_DESERIALIZATION_ERRORS =
//             "STUDENT_EVENT_DESERIALIZATION_ERRORS";
// }
// 
// Path: src/main/java/com/knewton/mapreduce/util/SerializationUtils.java
// public class SerializationUtils {
// 
//     public static final Class<Factory> SERIALIZATION_FACTORY_PARAMETER_DEFAULT =
//         TCompactProtocol.Factory.class;
// 
//     public static TDeserializer getDeserializerFromConf(Configuration conf) {
//         Class<? extends TProtocolFactory> protocolFactoryClass =
//             conf.getClass(PropertyConstants.SERIALIZATION_FACTORY_PARAMETER.txt,
//                           SERIALIZATION_FACTORY_PARAMETER_DEFAULT, TProtocolFactory.class);
//         TProtocolFactory protocolFactory = ReflectionUtils.newInstance(protocolFactoryClass, conf);
//         return new TDeserializer(protocolFactory);
//     }
// 
// }


import com.knewton.mapreduce.SSTableCellMapper;
import com.knewton.mapreduce.constant.PropertyConstants;
import com.knewton.mapreduce.util.CounterConstants;
import com.knewton.mapreduce.util.SerializationUtils;
import com.knewton.thrift.StudentEvent;
import com.knewton.thrift.StudentEventData;
import org.apache.cassandra.db.Cell;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Interval;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import javax.annotation.Nullable;
        studentEvent.setId(eventId);
        studentEvent.setData(studentEventData);
        context.getCounter(CounterConstants.STUDENT_EVENTS_JOB,
                CounterConstants.STUDENT_EVENTS_PROCESSED).increment(1);
        return studentEvent;
    }

    /**
     * Checks to see if the event is in the desired time range.
     *
     * @return True if the event is in the configured time interval
     */
    private boolean isInTimeRange(long eventId, Context context) {
        DateTime eventTime = new DateTime(eventId, DateTimeZone.UTC);
        // Skip events outside of the desired time range.
        if (timeRange != null && !timeRange.contains(eventTime)) {
            context.getCounter(CounterConstants.STUDENT_EVENTS_JOB,
                               CounterConstants.STUDENT_EVENTS_SKIPPED).increment(1);
            return false;
        }
        return true;
    }

    /**
     * Sets up a DateTime interval for excluding student events. When start time is not set then it
     * defaults to the beginning of time. If end date is not specified then it defaults to
     * "the end of time".
     */
    private void setupTimeRange(Configuration conf) {
        DateTimeFormatter dtf = DateTimeFormat.forPattern(DATE_TIME_STRING_FORMAT).withZoneUTC();
// Path: src/main/java/com/knewton/mapreduce/SSTableCellMapper.java
// public abstract class SSTableCellMapper<K1, V1,
//                                         K2 extends WritableComparable<?>, V2 extends Writable>
//         extends AbstractSSTableMapper<Cell, K1, V1, K2, V2> {
// 
//     public SSTableCellMapper() {
//         super(Cell.class);
//     }
// 
// }
// 
// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }
// 
// Path: src/main/java/com/knewton/mapreduce/util/CounterConstants.java
// public class CounterConstants {
//     public static final String STUDENT_EVENTS_JOB = "STUDENT_EVENTS";
//     public static final String STUDENT_EVENTS_SKIPPED = "STUDENT_EVENTS_SKIPPED";
//     public static final String STUDENT_EVENTS_COUNT = "STUDENT_EVENTS_COUNT";
//     public static final String STUDENT_EVENTS_PROCESSED = "STUDENT_EVENTS_PROCESSED";
//     public static final String STUDENT_EVENT_DESERIALIZATION_ERRORS =
//             "STUDENT_EVENT_DESERIALIZATION_ERRORS";
// }
// 
// Path: src/main/java/com/knewton/mapreduce/util/SerializationUtils.java
// public class SerializationUtils {
// 
//     public static final Class<Factory> SERIALIZATION_FACTORY_PARAMETER_DEFAULT =
//         TCompactProtocol.Factory.class;
// 
//     public static TDeserializer getDeserializerFromConf(Configuration conf) {
//         Class<? extends TProtocolFactory> protocolFactoryClass =
//             conf.getClass(PropertyConstants.SERIALIZATION_FACTORY_PARAMETER.txt,
//                           SERIALIZATION_FACTORY_PARAMETER_DEFAULT, TProtocolFactory.class);
//         TProtocolFactory protocolFactory = ReflectionUtils.newInstance(protocolFactoryClass, conf);
//         return new TDeserializer(protocolFactory);
//     }
// 
// }


// Path: src/main/java/com/knewton/mapreduce/example/StudentEventAbstractMapper.java
import com.knewton.mapreduce.SSTableCellMapper;
import com.knewton.mapreduce.constant.PropertyConstants;
import com.knewton.mapreduce.util.CounterConstants;
import com.knewton.mapreduce.util.SerializationUtils;
import com.knewton.thrift.StudentEvent;
import com.knewton.thrift.StudentEventData;
import org.apache.cassandra.db.Cell;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Interval;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.ByteBuffer;
import javax.annotation.Nullable;

        studentEvent.setId(eventId);
        studentEvent.setData(studentEventData);
        context.getCounter(CounterConstants.STUDENT_EVENTS_JOB,
                CounterConstants.STUDENT_EVENTS_PROCESSED).increment(1);
        return studentEvent;
    }

    /**
     * Checks to see if the event is in the desired time range.
     *
     * @return True if the event is in the configured time interval
     */
    private boolean isInTimeRange(long eventId, Context context) {
        DateTime eventTime = new DateTime(eventId, DateTimeZone.UTC);
        // Skip events outside of the desired time range.
        if (timeRange != null && !timeRange.contains(eventTime)) {
            context.getCounter(CounterConstants.STUDENT_EVENTS_JOB,
                               CounterConstants.STUDENT_EVENTS_SKIPPED).increment(1);
            return false;
        }
        return true;
    }

    /**
     * Sets up a DateTime interval for excluding student events. When start time is not set then it
     * defaults to the beginning of time. If end date is not specified then it defaults to
     * "the end of time".
     */
    private void setupTimeRange(Configuration conf) {
        DateTimeFormatter dtf = DateTimeFormat.forPattern(DATE_TIME_STRING_FORMAT).withZoneUTC();
        String startDateStr = conf.get(PropertyConstants.START_DATE.txt);