<commit_before>package com.backtype.hadoop.pail;

import com.backtype.hadoop.formats.RecordInputStream;
import com.backtype.hadoop.formats.RecordOutputStream;
import com.backtype.hadoop.formats.SequenceFileInputStream;
import com.backtype.hadoop.formats.SequenceFileOutputStream;
import com.backtype.support.KeywordArgParser;
import com.backtype.support.Utils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.SnappyCodec;
import org.apache.hadoop.mapred.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class SequenceFileFormat implements PailFormat {
    public static final String TYPE_ARG = "compressionType";
    public static final String CODEC_ARG = "compressionCodec";

    public static final String TYPE_ARG_NONE = "none";
    public static final String TYPE_ARG_RECORD = "record";
    public static final String TYPE_ARG_BLOCK = "block";

    public static final String CODEC_ARG_DEFAULT = "default";
    public static final String CODEC_ARG_GZIP = "gzip";
    public static final String CODEC_ARG_BZIP2 = "bzip2";
    public static final String CODEC_ARG_SNAPPY = "snappy";

    private static final Map<String, CompressionType> TYPES = new HashMap<String, CompressionType>() {{
        put(TYPE_ARG_RECORD, CompressionType.RECORD);
        put(TYPE_ARG_BLOCK, CompressionType.BLOCK);
    }};

    private static final Map<String, CompressionCodec> CODECS = new HashMap<String, CompressionCodec>() {{
        put(CODEC_ARG_DEFAULT, new DefaultCodec());
        put(CODEC_ARG_GZIP, new GzipCodec());
        put(CODEC_ARG_BZIP2, new BZip2Codec());
        put(CODEC_ARG_SNAPPY, new SnappyCodec());
    }};

    private String _typeArg;
    private String _codecArg;

    public SequenceFileFormat(Map<String, Object> args) {
        args = new KeywordArgParser()
                .add(TYPE_ARG, null, true, TYPE_ARG_RECORD, TYPE_ARG_BLOCK)
                .add(CODEC_ARG, CODEC_ARG_DEFAULT, false, CODEC_ARG_DEFAULT, CODEC_ARG_GZIP, CODEC_ARG_BZIP2, CODEC_ARG_SNAPPY)
                .parse(args);
        _typeArg = (String) args.get(TYPE_ARG);
        _codecArg = (String) args.get(CODEC_ARG);
    }

    public RecordInputStream getInputStream(FileSystem fs, Path path) throws IOException {
        return new SequenceFileInputStream(fs, path);
    }

    public RecordOutputStream getOutputStream(FileSystem fs, Path path) throws IOException {
        CompressionType type = TYPES.get(_typeArg);
        CompressionCodec codec = CODECS.get(_codecArg);

        if(type==null)
            return new SequenceFileOutputStream(fs, path);
        else
            return new SequenceFileOutputStream(fs, path, type, codec);
    }

    public Class<? extends InputFormat> getInputFormatClass() {
        return SequenceFilePailInputFormat.class;
    }


    public static class SequenceFilePailRecordReader implements RecordReader<Text, BytesWritable> {<commit_msg>Revert "Merge pull request #26 from fkautz/develop"<commit_after>package com.backtype.hadoop.pail;

import com.backtype.hadoop.formats.RecordInputStream;
import com.backtype.hadoop.formats.RecordOutputStream;
import com.backtype.hadoop.formats.SequenceFileInputStream;
import com.backtype.hadoop.formats.SequenceFileOutputStream;
import com.backtype.support.KeywordArgParser;
import com.backtype.support.Utils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class SequenceFileFormat implements PailFormat {
    public static final String TYPE_ARG = "compressionType";
    public static final String CODEC_ARG = "compressionCodec";

    public static final String TYPE_ARG_NONE = "none";
    public static final String TYPE_ARG_RECORD = "record";
    public static final String TYPE_ARG_BLOCK = "block";

    public static final String CODEC_ARG_DEFAULT = "default";
    public static final String CODEC_ARG_GZIP = "gzip";
    public static final String CODEC_ARG_BZIP2 = "bzip2";

    private static final Map<String, CompressionType> TYPES = new HashMap<String, CompressionType>() {{
        put(TYPE_ARG_RECORD, CompressionType.RECORD);
        put(TYPE_ARG_BLOCK, CompressionType.BLOCK);
    }};

    private static final Map<String, CompressionCodec> CODECS = new HashMap<String, CompressionCodec>() {{
        put(CODEC_ARG_DEFAULT, new DefaultCodec());
        put(CODEC_ARG_GZIP, new GzipCodec());
        put(CODEC_ARG_BZIP2, new BZip2Codec());
    }};

    private String _typeArg;
    private String _codecArg;

    public SequenceFileFormat(Map<String, Object> args) {
        args = new KeywordArgParser()
                .add(TYPE_ARG, null, true, TYPE_ARG_RECORD, TYPE_ARG_BLOCK)
                .add(CODEC_ARG, CODEC_ARG_DEFAULT, false, CODEC_ARG_DEFAULT, CODEC_ARG_GZIP, CODEC_ARG_BZIP2)
                .parse(args);
        _typeArg = (String) args.get(TYPE_ARG);
        _codecArg = (String) args.get(CODEC_ARG);
    }

    public RecordInputStream getInputStream(FileSystem fs, Path path) throws IOException {
        return new SequenceFileInputStream(fs, path);
    }

    public RecordOutputStream getOutputStream(FileSystem fs, Path path) throws IOException {
        CompressionType type = TYPES.get(_typeArg);
        CompressionCodec codec = CODECS.get(_codecArg);

        if(type==null)
            return new SequenceFileOutputStream(fs, path);
        else
            return new SequenceFileOutputStream(fs, path, type, codec);
    }

    public Class<? extends InputFormat> getInputFormatClass() {
        return SequenceFilePailInputFormat.class;
    }


    public static class SequenceFilePailRecordReader implements RecordReader<Text, BytesWritable> {
{'fraction_non_alphanumeric': 0.09257654534950896, 'fraction_numerical': 0.0025996533795493936, 'mean_word_length': 4.222473604826546, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6723149', 'n_tokens_mistral': 2086, 'n_tokens_neox': 2063, 'n_words': 460}