Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

HIVE-15273 adding confing to http client #119

Closed
wants to merge 1 commit into from
Closed

HIVE-15273 adding confing to http client #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 2 additions & 0 deletions common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1938,6 +1938,8 @@ public static enum ConfVars {
"number of records of the query results is larger than this threshold, we split the query in\n" +
"total number of rows/threshold parts across the time dimension. Note that we assume the\n" +
"records to be split uniformly across the time dimension"),
HIVE_DRUID_NUM_HTTP_CONNECTION("hive.druid.http.numConnection", 20, "Number of connections used by the HTTP client"),
HIVE_DRUID_HTTP_READ_TIMEOUT("hive.druid.http.read.timeout", "PT1M", "Read timeout period for the HTTP client in ISO8601 format, eg (P2W, P3M, PT1H30M, PT0.750S), default is period of 1 minute"),

// For HBase storage handler
HIVE_HBASE_WAL_ENABLED("hive.hbase.wal.enabled", true,
Expand Down
10 changes: 8 additions & 2 deletions druid-handler/src/java/org/apache/hadoop/hive/druid/HiveDruidQueryBasedInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.joda.time.Interval;
import org.joda.time.Period;
import org.joda.time.chrono.ISOChronology;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -160,7 +161,10 @@ private static HiveDruidSplit[] splitSelectQuery(Configuration conf, String addr
String druidQuery, Path dummyPath) throws IOException {
final int selectThreshold = (int) HiveConf.getIntVar(
conf, HiveConf.ConfVars.HIVE_DRUID_SELECT_THRESHOLD);

final int numConnection = HiveConf
.getIntVar(conf, HiveConf.ConfVars.HIVE_DRUID_NUM_HTTP_CONNECTION);
final Period readTimeout = new Period(
HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_DRUID_HTTP_READ_TIMEOUT));
SelectQuery query;
try {
query = DruidStorageHandlerUtils.JSON_MAPPER.readValue(druidQuery, SelectQuery.class);
Expand All @@ -184,7 +188,9 @@ private static HiveDruidSplit[] splitSelectQuery(Configuration conf, String addr
metadataBuilder.analysisTypes();
SegmentMetadataQuery metadataQuery = metadataBuilder.build();

HttpClient client = HttpClientInit.createClient(HttpClientConfig.builder().build(), new Lifecycle());
HttpClient client = HttpClientInit.createClient(
HttpClientConfig.builder().withNumConnections(numConnection)
.withReadTimeout(readTimeout.toStandardDuration()).build(), new Lifecycle());
InputStream response;
try {
response = DruidStorageHandlerUtils.submitRequest(client,
Expand Down
11 changes: 10 additions & 1 deletion druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,14 @@
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.druid.DruidStorageHandlerUtils;
import org.apache.hadoop.hive.druid.HiveDruidSplit;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.joda.time.Period;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -81,7 +83,14 @@ public void initialize(InputSplit split, Configuration conf) throws IOException
LOG.info("Retrieving from druid using query:\n " + query);
}

HttpClient client = HttpClientInit.createClient(HttpClientConfig.builder().build(), new Lifecycle());
final int numConnection = HiveConf
.getIntVar(conf, HiveConf.ConfVars.HIVE_DRUID_NUM_HTTP_CONNECTION);
final Period readTimeout = new Period(
HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_DRUID_HTTP_READ_TIMEOUT));

HttpClient client = HttpClientInit.createClient(
HttpClientConfig.builder().withReadTimeout(readTimeout.toStandardDuration())
.withNumConnections(numConnection).build(), new Lifecycle());
InputStream response = DruidStorageHandlerUtils.submitRequest(client,
DruidStorageHandlerUtils.createRequest(hiveDruidSplit.getAddress(), query));

Expand Down
12 changes: 11 additions & 1 deletion druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSerDe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.StringUtils;
import org.joda.time.Period;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -82,6 +83,9 @@ public class DruidSerDe extends AbstractSerDe {
private PrimitiveTypeInfo[] types;
private ObjectInspector inspector;

private int numConnection;

private Period readTimeout;

@Override
public void initialize(Configuration configuration, Properties properties) throws SerDeException {
Expand Down Expand Up @@ -113,6 +117,10 @@ public void initialize(Configuration configuration, Properties properties) throw
throw new SerDeException("Druid broker address not specified in configuration");
}

numConnection = HiveConf
.getIntVar(configuration, HiveConf.ConfVars.HIVE_DRUID_NUM_HTTP_CONNECTION);
readTimeout = new Period(
HiveConf.getVar(configuration, HiveConf.ConfVars.HIVE_DRUID_HTTP_READ_TIMEOUT));
// Infer schema
SegmentAnalysis schemaInfo;
try {
Expand Down Expand Up @@ -184,7 +192,9 @@ public void initialize(Configuration configuration, Properties properties) throw
/* Submits the request and returns */
protected SegmentAnalysis submitMetadataRequest(String address, SegmentMetadataQuery query)
throws SerDeException, IOException {
HttpClient client = HttpClientInit.createClient(HttpClientConfig.builder().build(), new Lifecycle());
HttpClient client = HttpClientInit.createClient(
HttpClientConfig.builder().withNumConnections(numConnection)
.withReadTimeout(readTimeout.toStandardDuration()).build(), new Lifecycle());
InputStream response;
try {
response = DruidStorageHandlerUtils.submitRequest(client,
Expand Down