// Path: dkpro-c4corpus-hadoop/src/main/java/de/tudarmstadt/ukp/dkpro/c4corpus/hadoop/io/WARCWritable.java
// public class WARCWritable
//         implements Writable
// {
// 
//     private WARCRecord record;
// 
//     /**
//      * Creates an empty writable (with a null record).
//      */
//     public WARCWritable()
//     {
//         this.record = null;
//     }
// 
//     /**
//      * Creates a writable wrapper around a given WARCRecord.
//      *
//      * @param record existing record
//      */
//     public WARCWritable(WARCRecord record)
//     {
//         this.record = record;
//     }
// 
//     /**
//      * Returns the record currently wrapped by this writable.
//      *
//      * @return current record
//      */
//     public WARCRecord getRecord()
//     {
//         return record;
//     }
// 
//     /**
//      * Updates the record held within this writable wrapper.
//      *
//      * @param record the record to be set
//      */
//     public void setRecord(WARCRecord record)
//     {
//         this.record = record;
//     }
// 
//     /**
//      * Appends the current record to a {@link DataOutput} stream.
//      */
//     @Override
//     public void write(DataOutput out)
//             throws IOException
//     {
//         if (record != null) {
//             record.write(out);
//         }
//     }
// 
//     /**
//      * Parses a {@link WARCRecord} out of a {@link DataInput} stream, and makes it the
//      * writable's current record.
//      *
//      * @throws IOException if the input is malformed
//      */
//     @Override
//     public void readFields(DataInput in)
//             throws IOException
//     {
//         record = new WARCRecord(in);
//     }
// }
// 
// Path: dkpro-c4corpus-hadoop/src/main/java/de/tudarmstadt/ukp/dkpro/c4corpus/hadoop/statistics/helper/TextLongCountingReducer.java
// public class TextLongCountingReducer
//         extends Reducer<Text, LongWritable, Text, LongWritable>
// {
//     @Override
//     protected void reduce(Text key, Iterable<LongWritable> values, Context context)
//             throws IOException, InterruptedException
//     {
//         long sum = 0;
//         for (LongWritable intWritable : values) {
//             sum += intWritable.get();
//         }
// 
//         context.write(key, new LongWritable(sum));
//     }
// }


import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.io.WARCInputFormat;
import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.io.WARCWritable;
import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.statistics.helper.TextLongCountingReducer;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
        FileInputFormat.addInputPaths(job, commaSeparatedInputFiles);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args)
            throws Exception
    {
        ToolRunner.run(new SimpleTextSearch(), args);
    }

    /**
     * Reads clean text (can contain "minimal html") and emits occurrences of given patterns
     */
    public static class TextSearchMapper
            extends Mapper<LongWritable, WARCWritable, Text, LongWritable>
    {
        private static final LongWritable ONE = new LongWritable(1L);

        private Pattern pattern;

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException
        {
            pattern = Pattern.compile(context.getConfiguration().get(MAPREDUCE_MAP_REGEX));
        }

        @Override
// Path: dkpro-c4corpus-hadoop/src/main/java/de/tudarmstadt/ukp/dkpro/c4corpus/hadoop/io/WARCWritable.java
// public class WARCWritable
//         implements Writable
// {
// 
//     private WARCRecord record;
// 
//     /**
//      * Creates an empty writable (with a null record).
//      */
//     public WARCWritable()
//     {
//         this.record = null;
//     }
// 
//     /**
//      * Creates a writable wrapper around a given WARCRecord.
//      *
//      * @param record existing record
//      */
//     public WARCWritable(WARCRecord record)
//     {
//         this.record = record;
//     }
// 
//     /**
//      * Returns the record currently wrapped by this writable.
//      *
//      * @return current record
//      */
//     public WARCRecord getRecord()
//     {
//         return record;
//     }
// 
//     /**
//      * Updates the record held within this writable wrapper.
//      *
//      * @param record the record to be set
//      */
//     public void setRecord(WARCRecord record)
//     {
//         this.record = record;
//     }
// 
//     /**
//      * Appends the current record to a {@link DataOutput} stream.
//      */
//     @Override
//     public void write(DataOutput out)
//             throws IOException
//     {
//         if (record != null) {
//             record.write(out);
//         }
//     }
// 
//     /**
//      * Parses a {@link WARCRecord} out of a {@link DataInput} stream, and makes it the
//      * writable's current record.
//      *
//      * @throws IOException if the input is malformed
//      */
//     @Override
//     public void readFields(DataInput in)
//             throws IOException
//     {
//         record = new WARCRecord(in);
//     }
// }
// 
// Path: dkpro-c4corpus-hadoop/src/main/java/de/tudarmstadt/ukp/dkpro/c4corpus/hadoop/statistics/helper/TextLongCountingReducer.java
// public class TextLongCountingReducer
//         extends Reducer<Text, LongWritable, Text, LongWritable>
// {
//     @Override
//     protected void reduce(Text key, Iterable<LongWritable> values, Context context)
//             throws IOException, InterruptedException
//     {
//         long sum = 0;
//         for (LongWritable intWritable : values) {
//             sum += intWritable.get();
//         }
// 
//         context.write(key, new LongWritable(sum));
//     }
// }


// Path: dkpro-c4corpus-hadoop/src/main/java/de/tudarmstadt/ukp/dkpro/c4corpus/hadoop/examples/SimpleTextSearch.java
import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.io.WARCInputFormat;
import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.io.WARCWritable;
import de.tudarmstadt.ukp.dkpro.c4corpus.hadoop.statistics.helper.TextLongCountingReducer;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

        FileInputFormat.addInputPaths(job, commaSeparatedInputFiles);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args)
            throws Exception
    {
        ToolRunner.run(new SimpleTextSearch(), args);
    }

    /**
     * Reads clean text (can contain "minimal html") and emits occurrences of given patterns
     */
    public static class TextSearchMapper
            extends Mapper<LongWritable, WARCWritable, Text, LongWritable>
    {
        private static final LongWritable ONE = new LongWritable(1L);

        private Pattern pattern;

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException
        {
            pattern = Pattern.compile(context.getConfiguration().get(MAPREDUCE_MAP_REGEX));
        }

        @Override
        protected void map(LongWritable key, WARCWritable value, Context context)