You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by na...@apache.org on 2009/07/27 22:43:09 UTC
svn commit: r798277 - in /hadoop/hive/trunk: CHANGES.txt
common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
conf/hive-default.xml
ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java
Author: namit
Date: Mon Jul 27 20:43:09 2009
New Revision: 798277
URL: http://svn.apache.org/viewvc?rev=798277&view=rev
Log:
HIVE-690. Fix script operator timeout bug
(Zheng Shao via namit)
Modified:
hadoop/hive/trunk/CHANGES.txt
hadoop/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hadoop/hive/trunk/conf/hive-default.xml
hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java
Modified: hadoop/hive/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/CHANGES.txt?rev=798277&r1=798276&r2=798277&view=diff
==============================================================================
--- hadoop/hive/trunk/CHANGES.txt (original)
+++ hadoop/hive/trunk/CHANGES.txt Mon Jul 27 20:43:09 2009
@@ -424,6 +424,9 @@
HIVE-673. Bug in handling of null partitions.
(Namit Jain via zshao)
+ HIVE-690. Fix script operator timeout bug
+ (Zheng Shao via namit)
+
Release 0.3.1 - Unreleased
INCOMPATIBLE CHANGES
Modified: hadoop/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=798277&r1=798276&r2=798277&view=diff
==============================================================================
--- hadoop/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hadoop/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Mon Jul 27 20:43:09 2009
@@ -112,6 +112,7 @@
// for hive script operator
HIVETABLENAME("hive.table.name", ""),
HIVEPARTITIONNAME("hive.partition.name", ""),
+ HIVESCRIPTAUTOPROGRESS("hive.script.auto.progress", false),
HIVEMAPREDMODE("hive.mapred.mode", "nonstrict"),
HIVEALIAS("hive.alias", ""),
HIVEMAPSIDEAGGREGATE("hive.map.aggr", "true"),
Modified: hadoop/hive/trunk/conf/hive-default.xml
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/conf/hive-default.xml?rev=798277&r1=798276&r2=798277&view=diff
==============================================================================
--- hadoop/hive/trunk/conf/hive-default.xml (original)
+++ hadoop/hive/trunk/conf/hive-default.xml Mon Jul 27 20:43:09 2009
@@ -326,4 +326,10 @@
<description>Size of merged files at the end of the job</description>
</property>
+<property>
+ <name>hive.script.auto.progress</name>
+ <value>false</value>
+ <description>Whether Hive Tranform/Map/Reduce Clause should automatically send progress information to TaskTracker to avoid the task getting killed because of inactivity. Hive sends progress information when the script is outputting to stderr. This option removes the need of periodically producing stderr messages, but users should be cautious because this may prevent infinite loops in the scripts to be killed by TaskTracker. </description>
+</property>
+
</configuration>
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java?rev=798277&r1=798276&r2=798277&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ScriptOperator.java Mon Jul 27 20:43:09 2009
@@ -229,22 +229,25 @@
(HiveConf.getIntVar(hconf, HiveConf.ConfVars.SCRIPTERRORLIMIT)),
"ErrorProcessor");
- /* Timer that reports every 5 minutes to the jobtracker. This ensures that even if
- the user script is not returning rows for greater than that duration, a progress
- report is sent to the tracker so that the tracker does not think that the job
- is dead.
- */
- Integer exp_interval = null;
- int exp_int;
- exp_interval = Integer.decode(hconf.get("mapred.tasktracker.expiry.interval"));
- if (exp_interval != null)
- exp_int = exp_interval.intValue() / 2;
- else
- exp_int = 300000;
-
- rpTimer = new Timer(true);
- rpTimer.scheduleAtFixedRate(new ReporterTask(reporter), 0, exp_interval);
-
+ if (HiveConf.getBoolVar(hconf, HiveConf.ConfVars.HIVESCRIPTAUTOPROGRESS)) {
+ /* Timer that reports every 5 minutes to the jobtracker. This ensures that even if
+ the user script is not returning rows for greater than that duration, a progress
+ report is sent to the tracker so that the tracker does not think that the job
+ is dead.
+ */
+ Integer expInterval = Integer.decode(hconf.get("mapred.tasktracker.expiry.interval"));
+ int notificationInterval;
+ if (expInterval != null) {
+ notificationInterval = expInterval.intValue() / 2;
+ } else {
+ // 5 minutes
+ notificationInterval = 5 * 60 * 1000;
+ }
+
+ rpTimer = new Timer(true);
+ rpTimer.scheduleAtFixedRate(new ReporterTask(reporter), 0, notificationInterval);
+ }
+
// initialize all children before starting the script
initializeChildren(hconf);
outThread.start();
@@ -339,21 +342,48 @@
}
}
+ /**
+ * The processor for stderr stream.
+ *
+ * TODO: In the future when we move to hadoop 0.18 and above, we should borrow the logic
+ * from HadoopStreaming: PipeMapRed.java MRErrorThread to support counters and status
+ * updates.
+ */
class ErrorStreamProcessor implements StreamProcessor {
private long bytesCopied = 0;
private long maxBytes;
+ private long lastReportTime;
+
public ErrorStreamProcessor (int maxBytes) {
this.maxBytes = (long)maxBytes;
+ lastReportTime = 0;
}
+
public void processLine(Text line) throws HiveException {
+
+ String stringLine = line.toString();
+
+ // Report progress for each stderr line, but no more frequently than once per minute.
+ long now = System.currentTimeMillis();
+ // reporter is a member variable of the Operator class.
+ if (now - lastReportTime > 60 * 1000 && reporter != null) {
+ lastReportTime = now;
+ reporter.progress();
+ }
+
if((maxBytes < 0) || (bytesCopied < maxBytes)) {
- System.err.println(line.toString());
+ System.err.println(stringLine);
}
+ if (bytesCopied < maxBytes && bytesCopied + line.getLength() >= maxBytes) {
+ System.err.println("Operator " + id + " " + getName()
+ + ": exceeding stderr limit of " + maxBytes + " bytes, will truncate stderr messages.");
+ }
bytesCopied += line.getLength();
}
public void close() {
}
+
}