<commit_before>package com.thinkbiganalytics.esbulkloader;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

/**
 * 
 * 
 */
public class ElasticsearchRecordWriter extends RecordWriter<Text, Text> {
    private static final int DEFAULT_BUFFER_SIZE = 100000;
    private static final String ESBL_BUFFER_SIZE = "esbl.buffer_size";
    private static final String DEFAULT_ES_PORT = "9200";
    private static final String DEFAULT_ES_HOST = "localhost";
    private static final String ESBL_PORT = "esbl.port";
    private static final String ESBL_HOST = "esbl.host";

    private static Log log = LogFactory.getLog(ElasticsearchRecordWriter.class);

    private PostMethod method;
    private final String apiUrl;
    private final int bufferSize;
    private final StringBuffer buffer ;

    public ElasticsearchRecordWriter(TaskAttemptContext context) {
	String host = context.getConfiguration().get(ESBL_HOST);
	String port = context.getConfiguration().get(ESBL_PORT);
	String size = context.getConfiguration().get(ESBL_BUFFER_SIZE);
	host = (null == host ? DEFAULT_ES_HOST : host);
	port = (null == port ? DEFAULT_ES_PORT : port);
	bufferSize = (null == size ? DEFAULT_BUFFER_SIZE : Integer.parseInt(size));
	buffer = new StringBuffer(bufferSize);
	apiUrl = String.format("http://%s:%s/_bulk", host, port);
    }

    @Override
    public void write(Text key, Text value) throws IOException,
	    InterruptedException {

	buffer.append(value.toString());
	log.info(value.toString());
	
	checkFlush();
    }

    @Override
    public void close(TaskAttemptContext arg0) throws IOException,
	    InterruptedException {
	flush();
	method.releaseConnection();
    }

    private void checkFlush() throws UnsupportedEncodingException {
	if (buffer.length()>this.bufferSize) {
	    flush();
	    buffer.setLength(0);
	}
    }
    
    private void flush() throws UnsupportedEncodingException {
	method = new PostMethod(apiUrl);
	method.setRequestEntity(new StringRequestEntity(buffer.toString(),
		"text/json", "UTF-8"));

	BufferedReader br = null;
	HttpClient client = new HttpClient();
	try {
	    int returnCode = client.executeMethod(method);

	    if (returnCode == HttpStatus.SC_NOT_IMPLEMENTED) {
		log.error(String.format(
			"The Post method is not implemented by this URI (%s)",
			apiUrl));
		// still consume the response body
		method.getResponseBodyAsString();
	    } else {
		br = new BufferedReader(new InputStreamReader(
			method.getResponseBodyAsStream()));
		// if (log.isInfoEnabled()) {
		String readLine;
		while (((readLine = br.readLine()) != null)) {
		    System.out.println(readLine);
		}
		// }
	    }
	} catch (Exception e) {
	    log.error(e);
	} finally {
	    if (br != null) {
		try {
		    br.close();
		} catch (Exception fe) {
		    log.error(fe);
		}
	    }
	}
    }

}<commit_msg>Add TcpNoDelay and quiet debug messages<commit_after>package com.thinkbiganalytics.esbulkloader;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.apache.commons.httpclient.params.HttpClientParams;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

/**
 * 
 * 
 */
public class ElasticsearchRecordWriter extends RecordWriter<Text, Text> {
    private static final int DEFAULT_BUFFER_SIZE = 100000;
    private static final String ESBL_BUFFER_SIZE = "esbl.buffer_size";
    private static final String DEFAULT_ES_PORT = "9200";
    private static final String DEFAULT_ES_HOST = "localhost";
    private static final String ESBL_PORT = "esbl.port";
    private static final String ESBL_HOST = "esbl.host";

    private static Log log = LogFactory.getLog(ElasticsearchRecordWriter.class);

    private PostMethod method;
    private final String apiUrl;
    private final int bufferSize;
    private final StringBuffer buffer;
    private String host;
    private int port;

    public ElasticsearchRecordWriter(TaskAttemptContext context) {
	host = context.getConfiguration().get(ESBL_HOST);
	String portString = context.getConfiguration().get(ESBL_PORT);
	String size = context.getConfiguration().get(ESBL_BUFFER_SIZE);
	host = (null == host ? DEFAULT_ES_HOST : host);
	port = Integer.parseInt((null == portString ? DEFAULT_ES_PORT : portString));
	bufferSize = (null == size ? DEFAULT_BUFFER_SIZE : Integer.parseInt(size));
	buffer = new StringBuffer(bufferSize);
	apiUrl = String.format("http://%s:%s/_bulk", host, port);
    }

    @Override
    public void write(Text key, Text value) throws IOException,
	    InterruptedException {

	buffer.append(value.toString());
	log.debug(value.toString());
	
	checkFlush();
    }

    @Override
    public void close(TaskAttemptContext arg0) throws IOException,
	    InterruptedException {
	flush();
	method.releaseConnection();
    }

    private void checkFlush() throws UnsupportedEncodingException {
	if (buffer.length()>this.bufferSize) {
	    flush();
	    buffer.setLength(0);
	}
    }
    
    private void flush() throws UnsupportedEncodingException {
	method = new PostMethod(apiUrl);

	method.setRequestEntity(new StringRequestEntity(buffer.toString(),
		"text/json", "UTF-8"));

	BufferedReader br = null;
	HttpClientParams params = new HttpClientParams();
	params.setBooleanParameter("http.tcp.nodelay", true);
	HttpClient client = new HttpClient();
	client.setParams(params);
	try {
	    
	    int returnCode = client.executeMethod(method);

	    if (returnCode == HttpStatus.SC_NOT_IMPLEMENTED) {
		log.error(String.format(
			"The Post method is not implemented by this URI (%s)",
			apiUrl));
		// still consume the response body
		method.getResponseBodyAsString();
	    } else {
		br = new BufferedReader(new InputStreamReader(
			method.getResponseBodyAsStream()));
		if (log.isDebugEnabled()) {
		    String readLine;
		    while (((readLine = br.readLine()) != null)) {
			System.out.println(readLine);
		    }
		}
	    }
	} catch (Exception e) {
	    log.error(e);
	} finally {
	    if (br != null) {
		try {
		    br.close();
		} catch (Exception fe) {
		    log.error(fe);
		}
	    }
	}
    }

}
{'fraction_non_alphanumeric': 0.11187214611872145, 'fraction_numerical': 0.0037100456621004564, 'mean_word_length': 4.203414996288047, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '5630921', 'n_tokens_mistral': 2165, 'n_tokens_neox': 2055, 'n_words': 477}