github/sqoop
5
0
Fork 0
mirror of https://github.com/apache/sqoop.git synced 2025-05-03 00:42:27 +08:00
Code Issues Projects Releases Wiki Activity

SQOOP-574. Create a Hadoop-2.0.0 build profile.

Browse Source 
(Cheolsoo Park via Jarek Jarcec Cecho)


git-svn-id: https://svn.apache.org/repos/asf/sqoop/trunk@1372798 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jarek Jarcec Cecho 2012-08-14 09:28:59 +00:00
parent 7e2df74795
commit cd0b181fa7
3 changed files with 31 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
build.xml
View File

@ -73,8 +73,18 @@
</then>
</elseif>
<elseif>
<equals arg1="${hadoopversion}" arg2="200" />
<then>
<property name="hadoop.version" value="2.0.0-alpha" />
<property name="hbase.version" value="0.92.0" />
<property name="zookeeper.version" value="3.4.2" />
<property name="hadoop.version.full" value="2.0.0-alpha" />
</then>
</elseif>
<else>
<fail message="Unrecognized hadoopversion. Can only be 20, 23, or 100." />
<fail message="Unrecognized hadoopversion. Can only be 20, 23, 100, or 200." />
</else>
</if>

18
ivy.xml
View File

@ -40,11 +40,13 @@ under the License.
<conf name="hadoop23" visibility="private" extends="common,runtime,hbase" />
<conf name="hadoop20" visibility="private" extends="common,runtime,hbase" />
<conf name="hadoop100" visibility="private" extends="common,runtime,hbase" />
<conf name="hadoop200" visibility="private" extends="common,runtime,hbase" />
<conf name="test" visibility="private" extends="common,runtime"/>
<conf name="hadoop23test" visibility="private" extends="test,hadoop23" />
<conf name="hadoop20test" visibility="private" extends="test,hadoop20" />
<conf name="hadoop100test" visibility="private" extends="test,hadoop100" />
<conf name="hadoop200test" visibility="private" extends="test,hadoop200" />
<!-- We don't redistribute everything we depend on (e.g., Hadoop itself);
anything which Hadoop itself also depends on, we do not ship.
@ -69,6 +71,22 @@ under the License.
</publications>
<dependencies>
<!-- Dependencies for Hadoop 2.0.0 -->
<dependency org="org.apache.hadoop" name="hadoop-common"
rev="${hadoop.version}" conf="hadoop200->default">
<artifact name="hadoop-common" type="jar" />
<artifact name="hadoop-common" type="jar" m:classifier="tests"/>
</dependency>
<dependency org="org.apache.hadoop" name="hadoop-hdfs"
rev="${hadoop.version}" conf="hadoop200->default">
<artifact name="hadoop-hdfs" type="jar" />
<artifact name="hadoop-hdfs" type="jar" m:classifier="tests"/>
</dependency>
<dependency org="org.apache.hadoop" name="hadoop-mapreduce-client-common"
rev="${hadoop.version}" conf="hadoop200->default"/>
<dependency org="org.apache.hadoop" name="hadoop-mapreduce-client-core"
rev="${hadoop.version}" conf="hadoop200->default"/>
<!-- Dependencies for Hadoop 0.23 -->
<dependency org="org.apache.hadoop" name="hadoop-common"
rev="${hadoop.version}" conf="hadoop23->default">

122
src/java/org/apache/sqoop/mapreduce/AutoProgressMapper.java
View File

@ -21,7 +21,6 @@
import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
/**
* Identity mapper that continuously reports progress via a background thread.
@ -32,25 +31,6 @@ public class AutoProgressMapper<KEYIN, VALIN, KEYOUT, VALOUT>
public static final Log LOG = LogFactory.getLog(
AutoProgressMapper.class.getName());
/**
* Total number of millis for which progress will be reported by the
* auto-progress thread. If this is zero, then the auto-progress thread will
* never voluntarily exit.
*/
private int maxProgressPeriod;
/**
* Number of milliseconds to sleep for between loop iterations. Must be less
* than report interval.
*/
private int sleepInterval;
/**
* Number of milliseconds between calls to Reporter.progress().
* Should be a multiple of the sleepInterval.
*/
private int reportInterval;
public static final String MAX_PROGRESS_PERIOD_KEY =
"sqoop.mapred.auto.progress.max";
public static final String SLEEP_INTERVAL_KEY =
@ -67,112 +47,14 @@ public class AutoProgressMapper<KEYIN, VALIN, KEYOUT, VALOUT>
// Disable max progress, by default.
public static final int DEFAULT_MAX_PROGRESS = 0;
private class ProgressThread extends Thread {
private volatile boolean keepGoing; // While this is true, thread runs.
private Context context;
private long startTimeMillis;
private long lastReportMillis;
public ProgressThread(final Context ctxt) {
this.context = ctxt;
this.keepGoing = true;
}
public void signalShutdown() {
this.keepGoing = false; // volatile update.
this.interrupt();
}
public void run() {
this.lastReportMillis = System.currentTimeMillis();
this.startTimeMillis = this.lastReportMillis;
final long MAX_PROGRESS = AutoProgressMapper.this.maxProgressPeriod;
final long REPORT_INTERVAL = AutoProgressMapper.this.reportInterval;
final long SLEEP_INTERVAL = AutoProgressMapper.this.sleepInterval;
// In a loop:
// * Check that we haven't run for too long (maxProgressPeriod).
// * If it's been a report interval since we last made progress,
// make more.
// * Sleep for a bit.
// * If the parent thread has signaled for exit, do so.
while (this.keepGoing) {
long curTimeMillis = System.currentTimeMillis();
if (MAX_PROGRESS != 0
&& curTimeMillis - this.startTimeMillis > MAX_PROGRESS) {
this.keepGoing = false;
LOG.info("Auto-progress thread exiting after " + MAX_PROGRESS
+ " ms.");
break;
}
if (curTimeMillis - this.lastReportMillis > REPORT_INTERVAL) {
// It's been a full report interval -- claim progress.
LOG.debug("Auto-progress thread reporting progress");
this.context.progress();
this.lastReportMillis = curTimeMillis;
}
// Unless we got an interrupt while we were working,
// sleep a bit before doing more work.
if (!Thread.interrupted()) {
try {
Thread.sleep(SLEEP_INTERVAL);
} catch (InterruptedException ie) {
// we were notified on something; not necessarily an error.
}
}
}
LOG.info("Auto-progress thread is finished. keepGoing=" + this.keepGoing);
}
}
/**
* Set configuration parameters for the auto-progress thread.
*/
private void configureAutoProgress(Configuration job) {
this.maxProgressPeriod = job.getInt(MAX_PROGRESS_PERIOD_KEY,
DEFAULT_MAX_PROGRESS);
this.sleepInterval = job.getInt(SLEEP_INTERVAL_KEY,
DEFAULT_SLEEP_INTERVAL);
this.reportInterval = job.getInt(REPORT_INTERVAL_KEY,
DEFAULT_REPORT_INTERVAL);
if (this.reportInterval < 1) {
LOG.warn("Invalid " + REPORT_INTERVAL_KEY + "; setting to "
+ DEFAULT_REPORT_INTERVAL);
this.reportInterval = DEFAULT_REPORT_INTERVAL;
}
if (this.sleepInterval > this.reportInterval || this.sleepInterval < 1) {
LOG.warn("Invalid " + SLEEP_INTERVAL_KEY + "; setting to "
+ DEFAULT_SLEEP_INTERVAL);
this.sleepInterval = DEFAULT_SLEEP_INTERVAL;
}
if (this.maxProgressPeriod < 0) {
LOG.warn("Invalid " + MAX_PROGRESS_PERIOD_KEY + "; setting to "
+ DEFAULT_MAX_PROGRESS);
this.maxProgressPeriod = DEFAULT_MAX_PROGRESS;
}
}
// map() method intentionally omitted; Mapper.map() is the identity mapper.
/**
* Run the mapping process for this task, wrapped in an auto-progress system.
*/
@Override
public void run(Context context) throws IOException, InterruptedException {
configureAutoProgress(context.getConfiguration());
ProgressThread thread = this.new ProgressThread(context);
ProgressThread thread = new ProgressThread(context, LOG);
try {
thread.setDaemon(true);
@ -191,7 +73,7 @@ public void run(Context context) throws IOException, InterruptedException {
LOG.debug("Progress thread shutdown detected.");
} catch (InterruptedException ie) {
LOG.warn("Interrupted when waiting on auto-progress thread: "
+ ie.toString());
+ ie.toString(), ie);
}
}
}