// Path: src/main/java/com/alexholmes/hadooputils/util/AvroFiles.java
// public final class AvroFiles {
//     private AvroFiles() {}
// 
//     /**
//      * Creates an avro container file.
//      *
//      * @param file The file to create.
//      * @param schema The schema for the records the file should contain.
//      * @param records The records to put in the file.
//      * @param <T> The (java) type of the avro records.
//      * @return The created file.
//      */
//     public static <T> File createFile(File file, Schema schema, T... records)
//             throws IOException {
//         DatumWriter<T> datumWriter = new GenericDatumWriter<T>(schema);
//         DataFileWriter<T> fileWriter = new DataFileWriter<T>(datumWriter);
//         fileWriter.create(schema, file);
//         for (T record : records) {
//             fileWriter.append(record);
//         }
//         fileWriter.close();
// 
//         return file;
//     }
// }


import com.alexholmes.hadooputils.util.AvroFiles;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.hadoop.io.AvroKeyValue;
import org.apache.avro.io.DatumReader;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyValueInputFormat;
import org.apache.avro.mapreduce.AvroKeyValueOutputFormat;
import org.apache.avro.specific.SpecificDatumReader;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import static org.junit.Assert.*;
/*
 * Copyright 2013 Alex Holmes
 *
 * Licensed 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 com.alexholmes.hadooputils.combine.avro.mapreduce;





/**
 */
public class CombineAvroKeyValueInputFormatTest {
    @Rule
    public TemporaryFolder mTempDir = new TemporaryFolder();

    /**
     * Creates an Avro file of <docid, text> pairs to use for test input:
     *
     * +-----+-----------------------+
     * | KEY | VALUE                 |
     * +-----+-----------------------+
     * | 1   | "apple banana carrot" |
     * | 2   | "apple banana"        |
     * | 3   | "apple"               |
     * +-----+-----------------------+
     *
     * @return The avro file.
     */
    private File createInputFile() throws IOException {
        Schema keyValueSchema = AvroKeyValue.getSchema(
                Schema.create(Schema.Type.INT), Schema.create(Schema.Type.STRING));

        AvroKeyValue<Integer, CharSequence> record1
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record1.setKey(1);
        record1.setValue("apple banana carrot");

        AvroKeyValue<Integer, CharSequence> record2
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record2.setKey(2);
        record2.setValue("apple banana");

        AvroKeyValue<Integer, CharSequence> record3
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record3.setKey(3);
        record3.setValue("apple");

// Path: src/main/java/com/alexholmes/hadooputils/util/AvroFiles.java
// public final class AvroFiles {
//     private AvroFiles() {}
// 
//     /**
//      * Creates an avro container file.
//      *
//      * @param file The file to create.
//      * @param schema The schema for the records the file should contain.
//      * @param records The records to put in the file.
//      * @param <T> The (java) type of the avro records.
//      * @return The created file.
//      */
//     public static <T> File createFile(File file, Schema schema, T... records)
//             throws IOException {
//         DatumWriter<T> datumWriter = new GenericDatumWriter<T>(schema);
//         DataFileWriter<T> fileWriter = new DataFileWriter<T>(datumWriter);
//         fileWriter.create(schema, file);
//         for (T record : records) {
//             fileWriter.append(record);
//         }
//         fileWriter.close();
// 
//         return file;
//     }
// }


// Path: src/test/java/com/alexholmes/hadooputils/combine/avro/mapreduce/CombineAvroKeyValueInputFormatTest.java
import com.alexholmes.hadooputils.util.AvroFiles;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.hadoop.io.AvroKeyValue;
import org.apache.avro.io.DatumReader;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyValueInputFormat;
import org.apache.avro.mapreduce.AvroKeyValueOutputFormat;
import org.apache.avro.specific.SpecificDatumReader;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import static org.junit.Assert.*;

/*
 * Copyright 2013 Alex Holmes
 *
 * Licensed 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 com.alexholmes.hadooputils.combine.avro.mapreduce;





/**
 */
public class CombineAvroKeyValueInputFormatTest {
    @Rule
    public TemporaryFolder mTempDir = new TemporaryFolder();

    /**
     * Creates an Avro file of <docid, text> pairs to use for test input:
     *
     * +-----+-----------------------+
     * | KEY | VALUE                 |
     * +-----+-----------------------+
     * | 1   | "apple banana carrot" |
     * | 2   | "apple banana"        |
     * | 3   | "apple"               |
     * +-----+-----------------------+
     *
     * @return The avro file.
     */
    private File createInputFile() throws IOException {
        Schema keyValueSchema = AvroKeyValue.getSchema(
                Schema.create(Schema.Type.INT), Schema.create(Schema.Type.STRING));

        AvroKeyValue<Integer, CharSequence> record1
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record1.setKey(1);
        record1.setValue("apple banana carrot");

        AvroKeyValue<Integer, CharSequence> record2
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record2.setKey(2);
        record2.setValue("apple banana");

        AvroKeyValue<Integer, CharSequence> record3
                = new AvroKeyValue<Integer, CharSequence>(new GenericData.Record(keyValueSchema));
        record3.setKey(3);
        record3.setValue("apple");

        return AvroFiles.createFile(new File(mTempDir.getRoot(), "inputKeyValues.avro"),