http://git-wip-us.apache.org/repos/asf/hbase-site/blob/09ea0d5f/apidocs/org/apache/hadoop/hbase/mapreduce/CopyTable.html ---------------------------------------------------------------------- diff --git a/apidocs/org/apache/hadoop/hbase/mapreduce/CopyTable.html b/apidocs/org/apache/hadoop/hbase/mapreduce/CopyTable.html index bc9720d..56ae45a 100644 --- a/apidocs/org/apache/hadoop/hbase/mapreduce/CopyTable.html +++ b/apidocs/org/apache/hadoop/hbase/mapreduce/CopyTable.html @@ -119,7 +119,7 @@ var activeTableTab = "activeTableTab"; <hr> <br> <pre>@InterfaceAudience.Public -public class <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.52">CopyTable</a> +public class <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.54">CopyTable</a> extends org.apache.hadoop.conf.Configured implements org.apache.hadoop.util.Tool</pre> <div class="block">Tool used to copy a table to another one which can be on a different setup. @@ -218,7 +218,7 @@ implements org.apache.hadoop.util.Tool</pre> <ul class="blockListLast"> <li class="blockList"> <h4>CopyTable</h4> -<pre>public <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.52">CopyTable</a>()</pre> +<pre>public <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.54">CopyTable</a>()</pre> </li> </ul> </li> @@ -235,7 +235,7 @@ implements org.apache.hadoop.util.Tool</pre> <ul class="blockList"> <li class="blockList"> <h4>createSubmittableJob</h4> -<pre>public org.apache.hadoop.mapreduce.Job <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.82">createSubmittableJob</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) +<pre>public org.apache.hadoop.mapreduce.Job <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.110">createSubmittableJob</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) throws <a href="https://docs.oracle.com/javase/8/docs/api/java/io/IOException.html?is-external=true"; title="class or interface in java.io">IOException</a></pre> <div class="block">Sets up the actual job.</div> <dl> @@ -254,7 +254,7 @@ implements org.apache.hadoop.util.Tool</pre> <ul class="blockList"> <li class="blockList"> <h4>main</h4> -<pre>public static void <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.356">main</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) +<pre>public static void <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.400">main</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) throws <a href="https://docs.oracle.com/javase/8/docs/api/java/lang/Exception.html?is-external=true"; title="class or interface in java.lang">Exception</a></pre> <div class="block">Main entry point.</div> <dl> @@ -271,7 +271,7 @@ implements org.apache.hadoop.util.Tool</pre> <ul class="blockListLast"> <li class="blockList"> <h4>run</h4> -<pre>public int <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.362">run</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) +<pre>public int <a href="../../../../../src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html#line.406">run</a>(<a href="https://docs.oracle.com/javase/8/docs/api/java/lang/String.html?is-external=true"; title="class or interface in java.lang">String</a>[] args) throws <a href="https://docs.oracle.com/javase/8/docs/api/java/lang/Exception.html?is-external=true"; title="class or interface in java.lang">Exception</a></pre> <dl> <dt><span class="overrideSpecifyLabel">Specified by:</span></dt>
http://git-wip-us.apache.org/repos/asf/hbase-site/blob/09ea0d5f/apidocs/src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html ---------------------------------------------------------------------- diff --git a/apidocs/src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html b/apidocs/src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html index e2ca07a..333ff45 100644 --- a/apidocs/src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html +++ b/apidocs/src-html/org/apache/hadoop/hbase/mapreduce/CopyTable.html @@ -29,7 +29,7 @@ <span class="sourceLineNo">021</span>import java.io.IOException;<a name="line.21"></a> <span class="sourceLineNo">022</span>import java.util.HashMap;<a name="line.22"></a> <span class="sourceLineNo">023</span>import java.util.Map;<a name="line.23"></a> -<span class="sourceLineNo">024</span>import java.util.Random;<a name="line.24"></a> +<span class="sourceLineNo">024</span>import java.util.UUID;<a name="line.24"></a> <span class="sourceLineNo">025</span><a name="line.25"></a> <span class="sourceLineNo">026</span>import org.apache.hadoop.conf.Configured;<a name="line.26"></a> <span class="sourceLineNo">027</span>import org.apache.hadoop.fs.FileSystem;<a name="line.27"></a> @@ -37,363 +37,410 @@ <span class="sourceLineNo">029</span>import org.apache.hadoop.hbase.HBaseConfiguration;<a name="line.29"></a> <span class="sourceLineNo">030</span>import org.apache.hadoop.hbase.HConstants;<a name="line.30"></a> <span class="sourceLineNo">031</span>import org.apache.hadoop.hbase.TableName;<a name="line.31"></a> -<span class="sourceLineNo">032</span>import org.apache.hadoop.hbase.util.FSUtils;<a name="line.32"></a> -<span class="sourceLineNo">033</span>import org.apache.yetus.audience.InterfaceAudience;<a name="line.33"></a> -<span class="sourceLineNo">034</span>import org.slf4j.Logger;<a name="line.34"></a> -<span class="sourceLineNo">035</span>import org.slf4j.LoggerFactory;<a name="line.35"></a> -<span class="sourceLineNo">036</span>import org.apache.hadoop.hbase.client.Admin;<a name="line.36"></a> -<span class="sourceLineNo">037</span>import org.apache.hadoop.hbase.client.Connection;<a name="line.37"></a> -<span class="sourceLineNo">038</span>import org.apache.hadoop.hbase.client.ConnectionFactory;<a name="line.38"></a> -<span class="sourceLineNo">039</span>import org.apache.hadoop.hbase.client.Scan;<a name="line.39"></a> -<span class="sourceLineNo">040</span>import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;<a name="line.40"></a> -<span class="sourceLineNo">041</span>import org.apache.hadoop.hbase.util.Bytes;<a name="line.41"></a> -<span class="sourceLineNo">042</span>import org.apache.hadoop.mapreduce.Job;<a name="line.42"></a> -<span class="sourceLineNo">043</span>import org.apache.hadoop.util.Tool;<a name="line.43"></a> -<span class="sourceLineNo">044</span>import org.apache.hadoop.util.ToolRunner;<a name="line.44"></a> -<span class="sourceLineNo">045</span><a name="line.45"></a> -<span class="sourceLineNo">046</span>/**<a name="line.46"></a> -<span class="sourceLineNo">047</span> * Tool used to copy a table to another one which can be on a different setup.<a name="line.47"></a> -<span class="sourceLineNo">048</span> * It is also configurable with a start and time as well as a specification<a name="line.48"></a> -<span class="sourceLineNo">049</span> * of the region server implementation if different from the local cluster.<a name="line.49"></a> -<span class="sourceLineNo">050</span> */<a name="line.50"></a> -<span class="sourceLineNo">051</span>@InterfaceAudience.Public<a name="line.51"></a> -<span class="sourceLineNo">052</span>public class CopyTable extends Configured implements Tool {<a name="line.52"></a> -<span class="sourceLineNo">053</span> private static final Logger LOG = LoggerFactory.getLogger(CopyTable.class);<a name="line.53"></a> -<span class="sourceLineNo">054</span><a name="line.54"></a> -<span class="sourceLineNo">055</span> final static String NAME = "copytable";<a name="line.55"></a> -<span class="sourceLineNo">056</span> long startTime = 0;<a name="line.56"></a> -<span class="sourceLineNo">057</span> long endTime = HConstants.LATEST_TIMESTAMP;<a name="line.57"></a> -<span class="sourceLineNo">058</span> int batch = Integer.MAX_VALUE;<a name="line.58"></a> -<span class="sourceLineNo">059</span> int cacheRow = -1;<a name="line.59"></a> -<span class="sourceLineNo">060</span> int versions = -1;<a name="line.60"></a> -<span class="sourceLineNo">061</span> String tableName = null;<a name="line.61"></a> -<span class="sourceLineNo">062</span> String startRow = null;<a name="line.62"></a> -<span class="sourceLineNo">063</span> String stopRow = null;<a name="line.63"></a> -<span class="sourceLineNo">064</span> String dstTableName = null;<a name="line.64"></a> -<span class="sourceLineNo">065</span> String peerAddress = null;<a name="line.65"></a> -<span class="sourceLineNo">066</span> String families = null;<a name="line.66"></a> -<span class="sourceLineNo">067</span> boolean allCells = false;<a name="line.67"></a> -<span class="sourceLineNo">068</span> static boolean shuffle = false;<a name="line.68"></a> -<span class="sourceLineNo">069</span><a name="line.69"></a> -<span class="sourceLineNo">070</span> boolean bulkload = false;<a name="line.70"></a> -<span class="sourceLineNo">071</span> Path bulkloadDir = null;<a name="line.71"></a> -<span class="sourceLineNo">072</span><a name="line.72"></a> -<span class="sourceLineNo">073</span> private final static String JOB_NAME_CONF_KEY = "mapreduce.job.name";<a name="line.73"></a> +<span class="sourceLineNo">032</span>import org.apache.hadoop.hbase.mapreduce.Import.CellImporter;<a name="line.32"></a> +<span class="sourceLineNo">033</span>import org.apache.hadoop.hbase.mapreduce.Import.Importer;<a name="line.33"></a> +<span class="sourceLineNo">034</span>import org.apache.hadoop.hbase.util.FSUtils;<a name="line.34"></a> +<span class="sourceLineNo">035</span>import org.apache.yetus.audience.InterfaceAudience;<a name="line.35"></a> +<span class="sourceLineNo">036</span>import org.slf4j.Logger;<a name="line.36"></a> +<span class="sourceLineNo">037</span>import org.slf4j.LoggerFactory;<a name="line.37"></a> +<span class="sourceLineNo">038</span>import org.apache.hadoop.hbase.client.Admin;<a name="line.38"></a> +<span class="sourceLineNo">039</span>import org.apache.hadoop.hbase.client.Connection;<a name="line.39"></a> +<span class="sourceLineNo">040</span>import org.apache.hadoop.hbase.client.ConnectionFactory;<a name="line.40"></a> +<span class="sourceLineNo">041</span>import org.apache.hadoop.hbase.client.Scan;<a name="line.41"></a> +<span class="sourceLineNo">042</span>import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;<a name="line.42"></a> +<span class="sourceLineNo">043</span>import org.apache.hadoop.hbase.util.Bytes;<a name="line.43"></a> +<span class="sourceLineNo">044</span>import org.apache.hadoop.mapreduce.Job;<a name="line.44"></a> +<span class="sourceLineNo">045</span>import org.apache.hadoop.util.Tool;<a name="line.45"></a> +<span class="sourceLineNo">046</span>import org.apache.hadoop.util.ToolRunner;<a name="line.46"></a> +<span class="sourceLineNo">047</span><a name="line.47"></a> +<span class="sourceLineNo">048</span>/**<a name="line.48"></a> +<span class="sourceLineNo">049</span> * Tool used to copy a table to another one which can be on a different setup.<a name="line.49"></a> +<span class="sourceLineNo">050</span> * It is also configurable with a start and time as well as a specification<a name="line.50"></a> +<span class="sourceLineNo">051</span> * of the region server implementation if different from the local cluster.<a name="line.51"></a> +<span class="sourceLineNo">052</span> */<a name="line.52"></a> +<span class="sourceLineNo">053</span>@InterfaceAudience.Public<a name="line.53"></a> +<span class="sourceLineNo">054</span>public class CopyTable extends Configured implements Tool {<a name="line.54"></a> +<span class="sourceLineNo">055</span> private static final Logger LOG = LoggerFactory.getLogger(CopyTable.class);<a name="line.55"></a> +<span class="sourceLineNo">056</span><a name="line.56"></a> +<span class="sourceLineNo">057</span> final static String NAME = "copytable";<a name="line.57"></a> +<span class="sourceLineNo">058</span> long startTime = 0;<a name="line.58"></a> +<span class="sourceLineNo">059</span> long endTime = HConstants.LATEST_TIMESTAMP;<a name="line.59"></a> +<span class="sourceLineNo">060</span> int batch = Integer.MAX_VALUE;<a name="line.60"></a> +<span class="sourceLineNo">061</span> int cacheRow = -1;<a name="line.61"></a> +<span class="sourceLineNo">062</span> int versions = -1;<a name="line.62"></a> +<span class="sourceLineNo">063</span> String tableName = null;<a name="line.63"></a> +<span class="sourceLineNo">064</span> String startRow = null;<a name="line.64"></a> +<span class="sourceLineNo">065</span> String stopRow = null;<a name="line.65"></a> +<span class="sourceLineNo">066</span> String dstTableName = null;<a name="line.66"></a> +<span class="sourceLineNo">067</span> String peerAddress = null;<a name="line.67"></a> +<span class="sourceLineNo">068</span> String families = null;<a name="line.68"></a> +<span class="sourceLineNo">069</span> boolean allCells = false;<a name="line.69"></a> +<span class="sourceLineNo">070</span> static boolean shuffle = false;<a name="line.70"></a> +<span class="sourceLineNo">071</span><a name="line.71"></a> +<span class="sourceLineNo">072</span> boolean bulkload = false;<a name="line.72"></a> +<span class="sourceLineNo">073</span> Path bulkloadDir = null;<a name="line.73"></a> <span class="sourceLineNo">074</span><a name="line.74"></a> -<span class="sourceLineNo">075</span> /**<a name="line.75"></a> -<span class="sourceLineNo">076</span> * Sets up the actual job.<a name="line.76"></a> -<span class="sourceLineNo">077</span> *<a name="line.77"></a> -<span class="sourceLineNo">078</span> * @param args The command line parameters.<a name="line.78"></a> -<span class="sourceLineNo">079</span> * @return The newly created job.<a name="line.79"></a> -<span class="sourceLineNo">080</span> * @throws IOException When setting up the job fails.<a name="line.80"></a> -<span class="sourceLineNo">081</span> */<a name="line.81"></a> -<span class="sourceLineNo">082</span> public Job createSubmittableJob(String[] args)<a name="line.82"></a> -<span class="sourceLineNo">083</span> throws IOException {<a name="line.83"></a> -<span class="sourceLineNo">084</span> if (!doCommandLine(args)) {<a name="line.84"></a> -<span class="sourceLineNo">085</span> return null;<a name="line.85"></a> -<span class="sourceLineNo">086</span> }<a name="line.86"></a> -<span class="sourceLineNo">087</span><a name="line.87"></a> -<span class="sourceLineNo">088</span> Job job = Job.getInstance(getConf(), getConf().get(JOB_NAME_CONF_KEY, NAME + "_" + tableName));<a name="line.88"></a> -<span class="sourceLineNo">089</span> job.setJarByClass(CopyTable.class);<a name="line.89"></a> -<span class="sourceLineNo">090</span> Scan scan = new Scan();<a name="line.90"></a> -<span class="sourceLineNo">091</span><a name="line.91"></a> -<span class="sourceLineNo">092</span> scan.setBatch(batch);<a name="line.92"></a> -<span class="sourceLineNo">093</span> scan.setCacheBlocks(false);<a name="line.93"></a> -<span class="sourceLineNo">094</span><a name="line.94"></a> -<span class="sourceLineNo">095</span> if (cacheRow > 0) {<a name="line.95"></a> -<span class="sourceLineNo">096</span> scan.setCaching(cacheRow);<a name="line.96"></a> -<span class="sourceLineNo">097</span> } else {<a name="line.97"></a> -<span class="sourceLineNo">098</span> scan.setCaching(getConf().getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, 100));<a name="line.98"></a> -<span class="sourceLineNo">099</span> }<a name="line.99"></a> -<span class="sourceLineNo">100</span><a name="line.100"></a> -<span class="sourceLineNo">101</span> scan.setTimeRange(startTime, endTime);<a name="line.101"></a> +<span class="sourceLineNo">075</span> boolean readingSnapshot = false;<a name="line.75"></a> +<span class="sourceLineNo">076</span> String snapshot = null;<a name="line.76"></a> +<span class="sourceLineNo">077</span><a name="line.77"></a> +<span class="sourceLineNo">078</span> private final static String JOB_NAME_CONF_KEY = "mapreduce.job.name";<a name="line.78"></a> +<span class="sourceLineNo">079</span><a name="line.79"></a> +<span class="sourceLineNo">080</span> private Path generateUniqTempDir(boolean withDirCreated) throws IOException {<a name="line.80"></a> +<span class="sourceLineNo">081</span> FileSystem fs = FSUtils.getCurrentFileSystem(getConf());<a name="line.81"></a> +<span class="sourceLineNo">082</span> Path dir = new Path(fs.getWorkingDirectory(), NAME);<a name="line.82"></a> +<span class="sourceLineNo">083</span> if (!fs.exists(dir)) {<a name="line.83"></a> +<span class="sourceLineNo">084</span> fs.mkdirs(dir);<a name="line.84"></a> +<span class="sourceLineNo">085</span> }<a name="line.85"></a> +<span class="sourceLineNo">086</span> Path newDir = new Path(dir, UUID.randomUUID().toString());<a name="line.86"></a> +<span class="sourceLineNo">087</span> if (withDirCreated) {<a name="line.87"></a> +<span class="sourceLineNo">088</span> fs.mkdirs(newDir);<a name="line.88"></a> +<span class="sourceLineNo">089</span> }<a name="line.89"></a> +<span class="sourceLineNo">090</span> return newDir;<a name="line.90"></a> +<span class="sourceLineNo">091</span> }<a name="line.91"></a> +<span class="sourceLineNo">092</span><a name="line.92"></a> +<span class="sourceLineNo">093</span> private void initCopyTableMapperReducerJob(Job job, Scan scan) throws IOException {<a name="line.93"></a> +<span class="sourceLineNo">094</span> Class<? extends TableMapper> mapper = bulkload ? CellImporter.class : Importer.class;<a name="line.94"></a> +<span class="sourceLineNo">095</span> if (readingSnapshot) {<a name="line.95"></a> +<span class="sourceLineNo">096</span> TableMapReduceUtil.initTableSnapshotMapperJob(snapshot, scan, mapper, null, null, job, true,<a name="line.96"></a> +<span class="sourceLineNo">097</span> generateUniqTempDir(true));<a name="line.97"></a> +<span class="sourceLineNo">098</span> } else {<a name="line.98"></a> +<span class="sourceLineNo">099</span> TableMapReduceUtil.initTableMapperJob(tableName, scan, mapper, null, null, job);<a name="line.99"></a> +<span class="sourceLineNo">100</span> }<a name="line.100"></a> +<span class="sourceLineNo">101</span> }<a name="line.101"></a> <span class="sourceLineNo">102</span><a name="line.102"></a> -<span class="sourceLineNo">103</span> if (allCells) {<a name="line.103"></a> -<span class="sourceLineNo">104</span> scan.setRaw(true);<a name="line.104"></a> -<span class="sourceLineNo">105</span> }<a name="line.105"></a> -<span class="sourceLineNo">106</span> if (shuffle) {<a name="line.106"></a> -<span class="sourceLineNo">107</span> job.getConfiguration().set(TableInputFormat.SHUFFLE_MAPS, "true");<a name="line.107"></a> -<span class="sourceLineNo">108</span> }<a name="line.108"></a> -<span class="sourceLineNo">109</span> if (versions >= 0) {<a name="line.109"></a> -<span class="sourceLineNo">110</span> scan.setMaxVersions(versions);<a name="line.110"></a> -<span class="sourceLineNo">111</span> }<a name="line.111"></a> -<span class="sourceLineNo">112</span><a name="line.112"></a> -<span class="sourceLineNo">113</span> if (startRow != null) {<a name="line.113"></a> -<span class="sourceLineNo">114</span> scan.setStartRow(Bytes.toBytesBinary(startRow));<a name="line.114"></a> -<span class="sourceLineNo">115</span> }<a name="line.115"></a> -<span class="sourceLineNo">116</span><a name="line.116"></a> -<span class="sourceLineNo">117</span> if (stopRow != null) {<a name="line.117"></a> -<span class="sourceLineNo">118</span> scan.setStopRow(Bytes.toBytesBinary(stopRow));<a name="line.118"></a> -<span class="sourceLineNo">119</span> }<a name="line.119"></a> -<span class="sourceLineNo">120</span><a name="line.120"></a> -<span class="sourceLineNo">121</span> if(families != null) {<a name="line.121"></a> -<span class="sourceLineNo">122</span> String[] fams = families.split(",");<a name="line.122"></a> -<span class="sourceLineNo">123</span> Map<String,String> cfRenameMap = new HashMap<>();<a name="line.123"></a> -<span class="sourceLineNo">124</span> for(String fam : fams) {<a name="line.124"></a> -<span class="sourceLineNo">125</span> String sourceCf;<a name="line.125"></a> -<span class="sourceLineNo">126</span> if(fam.contains(":")) {<a name="line.126"></a> -<span class="sourceLineNo">127</span> // fam looks like "sourceCfName:destCfName"<a name="line.127"></a> -<span class="sourceLineNo">128</span> String[] srcAndDest = fam.split(":", 2);<a name="line.128"></a> -<span class="sourceLineNo">129</span> sourceCf = srcAndDest[0];<a name="line.129"></a> -<span class="sourceLineNo">130</span> String destCf = srcAndDest[1];<a name="line.130"></a> -<span class="sourceLineNo">131</span> cfRenameMap.put(sourceCf, destCf);<a name="line.131"></a> -<span class="sourceLineNo">132</span> } else {<a name="line.132"></a> -<span class="sourceLineNo">133</span> // fam is just "sourceCf"<a name="line.133"></a> -<span class="sourceLineNo">134</span> sourceCf = fam;<a name="line.134"></a> -<span class="sourceLineNo">135</span> }<a name="line.135"></a> -<span class="sourceLineNo">136</span> scan.addFamily(Bytes.toBytes(sourceCf));<a name="line.136"></a> -<span class="sourceLineNo">137</span> }<a name="line.137"></a> -<span class="sourceLineNo">138</span> Import.configureCfRenaming(job.getConfiguration(), cfRenameMap);<a name="line.138"></a> +<span class="sourceLineNo">103</span> /**<a name="line.103"></a> +<span class="sourceLineNo">104</span> * Sets up the actual job.<a name="line.104"></a> +<span class="sourceLineNo">105</span> *<a name="line.105"></a> +<span class="sourceLineNo">106</span> * @param args The command line parameters.<a name="line.106"></a> +<span class="sourceLineNo">107</span> * @return The newly created job.<a name="line.107"></a> +<span class="sourceLineNo">108</span> * @throws IOException When setting up the job fails.<a name="line.108"></a> +<span class="sourceLineNo">109</span> */<a name="line.109"></a> +<span class="sourceLineNo">110</span> public Job createSubmittableJob(String[] args) throws IOException {<a name="line.110"></a> +<span class="sourceLineNo">111</span> if (!doCommandLine(args)) {<a name="line.111"></a> +<span class="sourceLineNo">112</span> return null;<a name="line.112"></a> +<span class="sourceLineNo">113</span> }<a name="line.113"></a> +<span class="sourceLineNo">114</span><a name="line.114"></a> +<span class="sourceLineNo">115</span> String jobName = NAME + "_" + (tableName == null ? snapshot : tableName);<a name="line.115"></a> +<span class="sourceLineNo">116</span> Job job = Job.getInstance(getConf(), getConf().get(JOB_NAME_CONF_KEY, jobName));<a name="line.116"></a> +<span class="sourceLineNo">117</span> job.setJarByClass(CopyTable.class);<a name="line.117"></a> +<span class="sourceLineNo">118</span> Scan scan = new Scan();<a name="line.118"></a> +<span class="sourceLineNo">119</span><a name="line.119"></a> +<span class="sourceLineNo">120</span> scan.setBatch(batch);<a name="line.120"></a> +<span class="sourceLineNo">121</span> scan.setCacheBlocks(false);<a name="line.121"></a> +<span class="sourceLineNo">122</span><a name="line.122"></a> +<span class="sourceLineNo">123</span> if (cacheRow > 0) {<a name="line.123"></a> +<span class="sourceLineNo">124</span> scan.setCaching(cacheRow);<a name="line.124"></a> +<span class="sourceLineNo">125</span> } else {<a name="line.125"></a> +<span class="sourceLineNo">126</span> scan.setCaching(getConf().getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, 100));<a name="line.126"></a> +<span class="sourceLineNo">127</span> }<a name="line.127"></a> +<span class="sourceLineNo">128</span><a name="line.128"></a> +<span class="sourceLineNo">129</span> scan.setTimeRange(startTime, endTime);<a name="line.129"></a> +<span class="sourceLineNo">130</span><a name="line.130"></a> +<span class="sourceLineNo">131</span> if (allCells) {<a name="line.131"></a> +<span class="sourceLineNo">132</span> scan.setRaw(true);<a name="line.132"></a> +<span class="sourceLineNo">133</span> }<a name="line.133"></a> +<span class="sourceLineNo">134</span> if (shuffle) {<a name="line.134"></a> +<span class="sourceLineNo">135</span> job.getConfiguration().set(TableInputFormat.SHUFFLE_MAPS, "true");<a name="line.135"></a> +<span class="sourceLineNo">136</span> }<a name="line.136"></a> +<span class="sourceLineNo">137</span> if (versions >= 0) {<a name="line.137"></a> +<span class="sourceLineNo">138</span> scan.readVersions(versions);<a name="line.138"></a> <span class="sourceLineNo">139</span> }<a name="line.139"></a> -<span class="sourceLineNo">140</span> job.setNumReduceTasks(0);<a name="line.140"></a> -<span class="sourceLineNo">141</span><a name="line.141"></a> -<span class="sourceLineNo">142</span> if (bulkload) {<a name="line.142"></a> -<span class="sourceLineNo">143</span> TableMapReduceUtil.initTableMapperJob(tableName, scan, Import.CellImporter.class, null, null,<a name="line.143"></a> -<span class="sourceLineNo">144</span> job);<a name="line.144"></a> -<span class="sourceLineNo">145</span><a name="line.145"></a> -<span class="sourceLineNo">146</span> // We need to split the inputs by destination tables so that output of Map can be bulk-loaded.<a name="line.146"></a> -<span class="sourceLineNo">147</span> TableInputFormat.configureSplitTable(job, TableName.valueOf(dstTableName));<a name="line.147"></a> +<span class="sourceLineNo">140</span><a name="line.140"></a> +<span class="sourceLineNo">141</span> if (startRow != null) {<a name="line.141"></a> +<span class="sourceLineNo">142</span> scan.withStartRow(Bytes.toBytesBinary(startRow));<a name="line.142"></a> +<span class="sourceLineNo">143</span> }<a name="line.143"></a> +<span class="sourceLineNo">144</span><a name="line.144"></a> +<span class="sourceLineNo">145</span> if (stopRow != null) {<a name="line.145"></a> +<span class="sourceLineNo">146</span> scan.withStopRow(Bytes.toBytesBinary(stopRow));<a name="line.146"></a> +<span class="sourceLineNo">147</span> }<a name="line.147"></a> <span class="sourceLineNo">148</span><a name="line.148"></a> -<span class="sourceLineNo">149</span> FileSystem fs = FSUtils.getCurrentFileSystem(getConf());<a name="line.149"></a> -<span class="sourceLineNo">150</span> Random rand = new Random();<a name="line.150"></a> -<span class="sourceLineNo">151</span> Path root = new Path(fs.getWorkingDirectory(), "copytable");<a name="line.151"></a> -<span class="sourceLineNo">152</span> fs.mkdirs(root);<a name="line.152"></a> -<span class="sourceLineNo">153</span> while (true) {<a name="line.153"></a> -<span class="sourceLineNo">154</span> bulkloadDir = new Path(root, "" + rand.nextLong());<a name="line.154"></a> -<span class="sourceLineNo">155</span> if (!fs.exists(bulkloadDir)) {<a name="line.155"></a> -<span class="sourceLineNo">156</span> break;<a name="line.156"></a> -<span class="sourceLineNo">157</span> }<a name="line.157"></a> -<span class="sourceLineNo">158</span> }<a name="line.158"></a> -<span class="sourceLineNo">159</span><a name="line.159"></a> -<span class="sourceLineNo">160</span> System.out.println("HFiles will be stored at " + this.bulkloadDir);<a name="line.160"></a> -<span class="sourceLineNo">161</span> HFileOutputFormat2.setOutputPath(job, bulkloadDir);<a name="line.161"></a> -<span class="sourceLineNo">162</span> try (Connection conn = ConnectionFactory.createConnection(getConf());<a name="line.162"></a> -<span class="sourceLineNo">163</span> Admin admin = conn.getAdmin()) {<a name="line.163"></a> -<span class="sourceLineNo">164</span> HFileOutputFormat2.configureIncrementalLoadMap(job,<a name="line.164"></a> -<span class="sourceLineNo">165</span> admin.getDescriptor((TableName.valueOf(dstTableName))));<a name="line.165"></a> -<span class="sourceLineNo">166</span> }<a name="line.166"></a> -<span class="sourceLineNo">167</span> } else {<a name="line.167"></a> -<span class="sourceLineNo">168</span> TableMapReduceUtil.initTableMapperJob(tableName, scan,<a name="line.168"></a> -<span class="sourceLineNo">169</span> Import.Importer.class, null, null, job);<a name="line.169"></a> -<span class="sourceLineNo">170</span><a name="line.170"></a> -<span class="sourceLineNo">171</span> TableMapReduceUtil.initTableReducerJob(dstTableName, null, job, null, peerAddress, null,<a name="line.171"></a> -<span class="sourceLineNo">172</span> null);<a name="line.172"></a> -<span class="sourceLineNo">173</span> }<a name="line.173"></a> -<span class="sourceLineNo">174</span><a name="line.174"></a> -<span class="sourceLineNo">175</span> return job;<a name="line.175"></a> -<span class="sourceLineNo">176</span> }<a name="line.176"></a> -<span class="sourceLineNo">177</span><a name="line.177"></a> -<span class="sourceLineNo">178</span> /*<a name="line.178"></a> -<span class="sourceLineNo">179</span> * @param errorMsg Error message. Can be null.<a name="line.179"></a> -<span class="sourceLineNo">180</span> */<a name="line.180"></a> -<span class="sourceLineNo">181</span> private static void printUsage(final String errorMsg) {<a name="line.181"></a> -<span class="sourceLineNo">182</span> if (errorMsg != null && errorMsg.length() > 0) {<a name="line.182"></a> -<span class="sourceLineNo">183</span> System.err.println("ERROR: " + errorMsg);<a name="line.183"></a> -<span class="sourceLineNo">184</span> }<a name="line.184"></a> -<span class="sourceLineNo">185</span> System.err.println("Usage: CopyTable [general options] [--starttime=X] [--endtime=Y] " +<a name="line.185"></a> -<span class="sourceLineNo">186</span> "[--new.name=NEW] [--peer.adr=ADR] <tablename>");<a name="line.186"></a> -<span class="sourceLineNo">187</span> System.err.println();<a name="line.187"></a> -<span class="sourceLineNo">188</span> System.err.println("Options:");<a name="line.188"></a> -<span class="sourceLineNo">189</span> System.err.println(" rs.class hbase.regionserver.class of the peer cluster");<a name="line.189"></a> -<span class="sourceLineNo">190</span> System.err.println(" specify if different from current cluster");<a name="line.190"></a> -<span class="sourceLineNo">191</span> System.err.println(" rs.impl hbase.regionserver.impl of the peer cluster");<a name="line.191"></a> -<span class="sourceLineNo">192</span> System.err.println(" startrow the start row");<a name="line.192"></a> -<span class="sourceLineNo">193</span> System.err.println(" stoprow the stop row");<a name="line.193"></a> -<span class="sourceLineNo">194</span> System.err.println(" starttime beginning of the time range (unixtime in millis)");<a name="line.194"></a> -<span class="sourceLineNo">195</span> System.err.println(" without endtime means from starttime to forever");<a name="line.195"></a> -<span class="sourceLineNo">196</span> System.err.println(" endtime end of the time range. Ignored if no starttime specified.");<a name="line.196"></a> -<span class="sourceLineNo">197</span> System.err.println(" versions number of cell versions to copy");<a name="line.197"></a> -<span class="sourceLineNo">198</span> System.err.println(" new.name new table's name");<a name="line.198"></a> -<span class="sourceLineNo">199</span> System.err.println(" peer.adr Address of the peer cluster given in the format");<a name="line.199"></a> -<span class="sourceLineNo">200</span> System.err.println(" hbase.zookeeper.quorum:hbase.zookeeper.client"<a name="line.200"></a> -<span class="sourceLineNo">201</span> + ".port:zookeeper.znode.parent");<a name="line.201"></a> -<span class="sourceLineNo">202</span> System.err.println(" families comma-separated list of families to copy");<a name="line.202"></a> -<span class="sourceLineNo">203</span> System.err.println(" To copy from cf1 to cf2, give sourceCfName:destCfName. ");<a name="line.203"></a> -<span class="sourceLineNo">204</span> System.err.println(" To keep the same name, just give \"cfName\"");<a name="line.204"></a> -<span class="sourceLineNo">205</span> System.err.println(" all.cells also copy delete markers and deleted cells");<a name="line.205"></a> -<span class="sourceLineNo">206</span> System.err.println(" bulkload Write input into HFiles and bulk load to the destination "<a name="line.206"></a> -<span class="sourceLineNo">207</span> + "table");<a name="line.207"></a> -<span class="sourceLineNo">208</span> System.err.println();<a name="line.208"></a> -<span class="sourceLineNo">209</span> System.err.println("Args:");<a name="line.209"></a> -<span class="sourceLineNo">210</span> System.err.println(" tablename Name of the table to copy");<a name="line.210"></a> -<span class="sourceLineNo">211</span> System.err.println();<a name="line.211"></a> -<span class="sourceLineNo">212</span> System.err.println("Examples:");<a name="line.212"></a> -<span class="sourceLineNo">213</span> System.err.println(" To copy 'TestTable' to a cluster that uses replication for a 1 hour window:");<a name="line.213"></a> -<span class="sourceLineNo">214</span> System.err.println(" $ hbase " +<a name="line.214"></a> -<span class="sourceLineNo">215</span> "org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1265875194289 --endtime=1265878794289 " +<a name="line.215"></a> -<span class="sourceLineNo">216</span> "--peer.adr=server1,server2,server3:2181:/hbase --families=myOldCf:myNewCf,cf2,cf3 TestTable ");<a name="line.216"></a> -<span class="sourceLineNo">217</span> System.err.println("For performance consider the following general option:\n"<a name="line.217"></a> -<span class="sourceLineNo">218</span> + " It is recommended that you set the following to >=100. A higher value uses more memory but\n"<a name="line.218"></a> -<span class="sourceLineNo">219</span> + " decreases the round trip time to the server and may increase performance.\n"<a name="line.219"></a> -<span class="sourceLineNo">220</span> + " -Dhbase.client.scanner.caching=100\n"<a name="line.220"></a> -<span class="sourceLineNo">221</span> + " The following should always be set to false, to prevent writing data twice, which may produce \n"<a name="line.221"></a> -<span class="sourceLineNo">222</span> + " inaccurate results.\n"<a name="line.222"></a> -<span class="sourceLineNo">223</span> + " -Dmapreduce.map.speculative=false");<a name="line.223"></a> -<span class="sourceLineNo">224</span> }<a name="line.224"></a> -<span class="sourceLineNo">225</span><a name="line.225"></a> -<span class="sourceLineNo">226</span> private boolean doCommandLine(final String[] args) {<a name="line.226"></a> -<span class="sourceLineNo">227</span> // Process command-line args. TODO: Better cmd-line processing<a name="line.227"></a> -<span class="sourceLineNo">228</span> // (but hopefully something not as painful as cli options).<a name="line.228"></a> -<span class="sourceLineNo">229</span> if (args.length < 1) {<a name="line.229"></a> -<span class="sourceLineNo">230</span> printUsage(null);<a name="line.230"></a> -<span class="sourceLineNo">231</span> return false;<a name="line.231"></a> -<span class="sourceLineNo">232</span> }<a name="line.232"></a> -<span class="sourceLineNo">233</span> try {<a name="line.233"></a> -<span class="sourceLineNo">234</span> for (int i = 0; i < args.length; i++) {<a name="line.234"></a> -<span class="sourceLineNo">235</span> String cmd = args[i];<a name="line.235"></a> -<span class="sourceLineNo">236</span> if (cmd.equals("-h") || cmd.startsWith("--h")) {<a name="line.236"></a> -<span class="sourceLineNo">237</span> printUsage(null);<a name="line.237"></a> -<span class="sourceLineNo">238</span> return false;<a name="line.238"></a> -<span class="sourceLineNo">239</span> }<a name="line.239"></a> -<span class="sourceLineNo">240</span><a name="line.240"></a> -<span class="sourceLineNo">241</span> final String startRowArgKey = "--startrow=";<a name="line.241"></a> -<span class="sourceLineNo">242</span> if (cmd.startsWith(startRowArgKey)) {<a name="line.242"></a> -<span class="sourceLineNo">243</span> startRow = cmd.substring(startRowArgKey.length());<a name="line.243"></a> -<span class="sourceLineNo">244</span> continue;<a name="line.244"></a> -<span class="sourceLineNo">245</span> }<a name="line.245"></a> -<span class="sourceLineNo">246</span><a name="line.246"></a> -<span class="sourceLineNo">247</span> final String stopRowArgKey = "--stoprow=";<a name="line.247"></a> -<span class="sourceLineNo">248</span> if (cmd.startsWith(stopRowArgKey)) {<a name="line.248"></a> -<span class="sourceLineNo">249</span> stopRow = cmd.substring(stopRowArgKey.length());<a name="line.249"></a> -<span class="sourceLineNo">250</span> continue;<a name="line.250"></a> -<span class="sourceLineNo">251</span> }<a name="line.251"></a> -<span class="sourceLineNo">252</span><a name="line.252"></a> -<span class="sourceLineNo">253</span> final String startTimeArgKey = "--starttime=";<a name="line.253"></a> -<span class="sourceLineNo">254</span> if (cmd.startsWith(startTimeArgKey)) {<a name="line.254"></a> -<span class="sourceLineNo">255</span> startTime = Long.parseLong(cmd.substring(startTimeArgKey.length()));<a name="line.255"></a> -<span class="sourceLineNo">256</span> continue;<a name="line.256"></a> -<span class="sourceLineNo">257</span> }<a name="line.257"></a> -<span class="sourceLineNo">258</span><a name="line.258"></a> -<span class="sourceLineNo">259</span> final String endTimeArgKey = "--endtime=";<a name="line.259"></a> -<span class="sourceLineNo">260</span> if (cmd.startsWith(endTimeArgKey)) {<a name="line.260"></a> -<span class="sourceLineNo">261</span> endTime = Long.parseLong(cmd.substring(endTimeArgKey.length()));<a name="line.261"></a> -<span class="sourceLineNo">262</span> continue;<a name="line.262"></a> -<span class="sourceLineNo">263</span> }<a name="line.263"></a> -<span class="sourceLineNo">264</span><a name="line.264"></a> -<span class="sourceLineNo">265</span> final String batchArgKey = "--batch=";<a name="line.265"></a> -<span class="sourceLineNo">266</span> if (cmd.startsWith(batchArgKey)) {<a name="line.266"></a> -<span class="sourceLineNo">267</span> batch = Integer.parseInt(cmd.substring(batchArgKey.length()));<a name="line.267"></a> -<span class="sourceLineNo">268</span> continue;<a name="line.268"></a> -<span class="sourceLineNo">269</span> }<a name="line.269"></a> -<span class="sourceLineNo">270</span><a name="line.270"></a> -<span class="sourceLineNo">271</span> final String cacheRowArgKey = "--cacheRow=";<a name="line.271"></a> -<span class="sourceLineNo">272</span> if (cmd.startsWith(cacheRowArgKey)) {<a name="line.272"></a> -<span class="sourceLineNo">273</span> cacheRow = Integer.parseInt(cmd.substring(cacheRowArgKey.length()));<a name="line.273"></a> -<span class="sourceLineNo">274</span> continue;<a name="line.274"></a> -<span class="sourceLineNo">275</span> }<a name="line.275"></a> -<span class="sourceLineNo">276</span><a name="line.276"></a> -<span class="sourceLineNo">277</span> final String versionsArgKey = "--versions=";<a name="line.277"></a> -<span class="sourceLineNo">278</span> if (cmd.startsWith(versionsArgKey)) {<a name="line.278"></a> -<span class="sourceLineNo">279</span> versions = Integer.parseInt(cmd.substring(versionsArgKey.length()));<a name="line.279"></a> -<span class="sourceLineNo">280</span> continue;<a name="line.280"></a> -<span class="sourceLineNo">281</span> }<a name="line.281"></a> -<span class="sourceLineNo">282</span><a name="line.282"></a> -<span class="sourceLineNo">283</span> final String newNameArgKey = "--new.name=";<a name="line.283"></a> -<span class="sourceLineNo">284</span> if (cmd.startsWith(newNameArgKey)) {<a name="line.284"></a> -<span class="sourceLineNo">285</span> dstTableName = cmd.substring(newNameArgKey.length());<a name="line.285"></a> -<span class="sourceLineNo">286</span> continue;<a name="line.286"></a> -<span class="sourceLineNo">287</span> }<a name="line.287"></a> -<span class="sourceLineNo">288</span><a name="line.288"></a> -<span class="sourceLineNo">289</span> final String peerAdrArgKey = "--peer.adr=";<a name="line.289"></a> -<span class="sourceLineNo">290</span> if (cmd.startsWith(peerAdrArgKey)) {<a name="line.290"></a> -<span class="sourceLineNo">291</span> peerAddress = cmd.substring(peerAdrArgKey.length());<a name="line.291"></a> -<span class="sourceLineNo">292</span> continue;<a name="line.292"></a> -<span class="sourceLineNo">293</span> }<a name="line.293"></a> -<span class="sourceLineNo">294</span><a name="line.294"></a> -<span class="sourceLineNo">295</span> final String familiesArgKey = "--families=";<a name="line.295"></a> -<span class="sourceLineNo">296</span> if (cmd.startsWith(familiesArgKey)) {<a name="line.296"></a> -<span class="sourceLineNo">297</span> families = cmd.substring(familiesArgKey.length());<a name="line.297"></a> -<span class="sourceLineNo">298</span> continue;<a name="line.298"></a> -<span class="sourceLineNo">299</span> }<a name="line.299"></a> -<span class="sourceLineNo">300</span><a name="line.300"></a> -<span class="sourceLineNo">301</span> if (cmd.startsWith("--all.cells")) {<a name="line.301"></a> -<span class="sourceLineNo">302</span> allCells = true;<a name="line.302"></a> -<span class="sourceLineNo">303</span> continue;<a name="line.303"></a> -<span class="sourceLineNo">304</span> }<a name="line.304"></a> -<span class="sourceLineNo">305</span><a name="line.305"></a> -<span class="sourceLineNo">306</span> if (cmd.startsWith("--bulkload")) {<a name="line.306"></a> -<span class="sourceLineNo">307</span> bulkload = true;<a name="line.307"></a> -<span class="sourceLineNo">308</span> continue;<a name="line.308"></a> -<span class="sourceLineNo">309</span> }<a name="line.309"></a> -<span class="sourceLineNo">310</span><a name="line.310"></a> -<span class="sourceLineNo">311</span> if (cmd.startsWith("--shuffle")) {<a name="line.311"></a> -<span class="sourceLineNo">312</span> shuffle = true;<a name="line.312"></a> -<span class="sourceLineNo">313</span> continue;<a name="line.313"></a> -<span class="sourceLineNo">314</span> }<a name="line.314"></a> -<span class="sourceLineNo">315</span><a name="line.315"></a> -<span class="sourceLineNo">316</span> if (i == args.length-1) {<a name="line.316"></a> -<span class="sourceLineNo">317</span> tableName = cmd;<a name="line.317"></a> -<span class="sourceLineNo">318</span> } else {<a name="line.318"></a> -<span class="sourceLineNo">319</span> printUsage("Invalid argument '" + cmd + "'");<a name="line.319"></a> -<span class="sourceLineNo">320</span> return false;<a name="line.320"></a> -<span class="sourceLineNo">321</span> }<a name="line.321"></a> -<span class="sourceLineNo">322</span> }<a name="line.322"></a> -<span class="sourceLineNo">323</span> if (dstTableName == null && peerAddress == null) {<a name="line.323"></a> -<span class="sourceLineNo">324</span> printUsage("At least a new table name or a " +<a name="line.324"></a> -<span class="sourceLineNo">325</span> "peer address must be specified");<a name="line.325"></a> -<span class="sourceLineNo">326</span> return false;<a name="line.326"></a> -<span class="sourceLineNo">327</span> }<a name="line.327"></a> -<span class="sourceLineNo">328</span> if ((endTime != 0) && (startTime > endTime)) {<a name="line.328"></a> -<span class="sourceLineNo">329</span> printUsage("Invalid time range filter: starttime=" + startTime + " > endtime=" + endTime);<a name="line.329"></a> -<span class="sourceLineNo">330</span> return false;<a name="line.330"></a> -<span class="sourceLineNo">331</span> }<a name="line.331"></a> -<span class="sourceLineNo">332</span><a name="line.332"></a> -<span class="sourceLineNo">333</span> if (bulkload && peerAddress != null) {<a name="line.333"></a> -<span class="sourceLineNo">334</span> printUsage("Remote bulkload is not supported!");<a name="line.334"></a> -<span class="sourceLineNo">335</span> return false;<a name="line.335"></a> -<span class="sourceLineNo">336</span> }<a name="line.336"></a> -<span class="sourceLineNo">337</span><a name="line.337"></a> -<span class="sourceLineNo">338</span> // set dstTableName if necessary<a name="line.338"></a> -<span class="sourceLineNo">339</span> if (dstTableName == null) {<a name="line.339"></a> -<span class="sourceLineNo">340</span> dstTableName = tableName;<a name="line.340"></a> -<span class="sourceLineNo">341</span> }<a name="line.341"></a> -<span class="sourceLineNo">342</span> } catch (Exception e) {<a name="line.342"></a> -<span class="sourceLineNo">343</span> LOG.error("Failed to parse commandLine arguments", e);<a name="line.343"></a> -<span class="sourceLineNo">344</span> printUsage("Can't start because " + e.getMessage());<a name="line.344"></a> -<span class="sourceLineNo">345</span> return false;<a name="line.345"></a> -<span class="sourceLineNo">346</span> }<a name="line.346"></a> -<span class="sourceLineNo">347</span> return true;<a name="line.347"></a> -<span class="sourceLineNo">348</span> }<a name="line.348"></a> -<span class="sourceLineNo">349</span><a name="line.349"></a> -<span class="sourceLineNo">350</span> /**<a name="line.350"></a> -<span class="sourceLineNo">351</span> * Main entry point.<a name="line.351"></a> -<span class="sourceLineNo">352</span> *<a name="line.352"></a> -<span class="sourceLineNo">353</span> * @param args The command line parameters.<a name="line.353"></a> -<span class="sourceLineNo">354</span> * @throws Exception When running the job fails.<a name="line.354"></a> -<span class="sourceLineNo">355</span> */<a name="line.355"></a> -<span class="sourceLineNo">356</span> public static void main(String[] args) throws Exception {<a name="line.356"></a> -<span class="sourceLineNo">357</span> int ret = ToolRunner.run(HBaseConfiguration.create(), new CopyTable(), args);<a name="line.357"></a> -<span class="sourceLineNo">358</span> System.exit(ret);<a name="line.358"></a> -<span class="sourceLineNo">359</span> }<a name="line.359"></a> +<span class="sourceLineNo">149</span> if(families != null) {<a name="line.149"></a> +<span class="sourceLineNo">150</span> String[] fams = families.split(",");<a name="line.150"></a> +<span class="sourceLineNo">151</span> Map<String,String> cfRenameMap = new HashMap<>();<a name="line.151"></a> +<span class="sourceLineNo">152</span> for(String fam : fams) {<a name="line.152"></a> +<span class="sourceLineNo">153</span> String sourceCf;<a name="line.153"></a> +<span class="sourceLineNo">154</span> if(fam.contains(":")) {<a name="line.154"></a> +<span class="sourceLineNo">155</span> // fam looks like "sourceCfName:destCfName"<a name="line.155"></a> +<span class="sourceLineNo">156</span> String[] srcAndDest = fam.split(":", 2);<a name="line.156"></a> +<span class="sourceLineNo">157</span> sourceCf = srcAndDest[0];<a name="line.157"></a> +<span class="sourceLineNo">158</span> String destCf = srcAndDest[1];<a name="line.158"></a> +<span class="sourceLineNo">159</span> cfRenameMap.put(sourceCf, destCf);<a name="line.159"></a> +<span class="sourceLineNo">160</span> } else {<a name="line.160"></a> +<span class="sourceLineNo">161</span> // fam is just "sourceCf"<a name="line.161"></a> +<span class="sourceLineNo">162</span> sourceCf = fam;<a name="line.162"></a> +<span class="sourceLineNo">163</span> }<a name="line.163"></a> +<span class="sourceLineNo">164</span> scan.addFamily(Bytes.toBytes(sourceCf));<a name="line.164"></a> +<span class="sourceLineNo">165</span> }<a name="line.165"></a> +<span class="sourceLineNo">166</span> Import.configureCfRenaming(job.getConfiguration(), cfRenameMap);<a name="line.166"></a> +<span class="sourceLineNo">167</span> }<a name="line.167"></a> +<span class="sourceLineNo">168</span> job.setNumReduceTasks(0);<a name="line.168"></a> +<span class="sourceLineNo">169</span><a name="line.169"></a> +<span class="sourceLineNo">170</span> if (bulkload) {<a name="line.170"></a> +<span class="sourceLineNo">171</span> initCopyTableMapperReducerJob(job, scan);<a name="line.171"></a> +<span class="sourceLineNo">172</span><a name="line.172"></a> +<span class="sourceLineNo">173</span> // We need to split the inputs by destination tables so that output of Map can be bulk-loaded.<a name="line.173"></a> +<span class="sourceLineNo">174</span> TableInputFormat.configureSplitTable(job, TableName.valueOf(dstTableName));<a name="line.174"></a> +<span class="sourceLineNo">175</span><a name="line.175"></a> +<span class="sourceLineNo">176</span> bulkloadDir = generateUniqTempDir(false);<a name="line.176"></a> +<span class="sourceLineNo">177</span> LOG.info("HFiles will be stored at " + this.bulkloadDir);<a name="line.177"></a> +<span class="sourceLineNo">178</span> HFileOutputFormat2.setOutputPath(job, bulkloadDir);<a name="line.178"></a> +<span class="sourceLineNo">179</span> try (Connection conn = ConnectionFactory.createConnection(getConf());<a name="line.179"></a> +<span class="sourceLineNo">180</span> Admin admin = conn.getAdmin()) {<a name="line.180"></a> +<span class="sourceLineNo">181</span> HFileOutputFormat2.configureIncrementalLoadMap(job,<a name="line.181"></a> +<span class="sourceLineNo">182</span> admin.getDescriptor((TableName.valueOf(dstTableName))));<a name="line.182"></a> +<span class="sourceLineNo">183</span> }<a name="line.183"></a> +<span class="sourceLineNo">184</span> } else {<a name="line.184"></a> +<span class="sourceLineNo">185</span> initCopyTableMapperReducerJob(job, scan);<a name="line.185"></a> +<span class="sourceLineNo">186</span> TableMapReduceUtil.initTableReducerJob(dstTableName, null, job, null, peerAddress, null,<a name="line.186"></a> +<span class="sourceLineNo">187</span> null);<a name="line.187"></a> +<span class="sourceLineNo">188</span> }<a name="line.188"></a> +<span class="sourceLineNo">189</span><a name="line.189"></a> +<span class="sourceLineNo">190</span> return job;<a name="line.190"></a> +<span class="sourceLineNo">191</span> }<a name="line.191"></a> +<span class="sourceLineNo">192</span><a name="line.192"></a> +<span class="sourceLineNo">193</span> /*<a name="line.193"></a> +<span class="sourceLineNo">194</span> * @param errorMsg Error message. Can be null.<a name="line.194"></a> +<span class="sourceLineNo">195</span> */<a name="line.195"></a> +<span class="sourceLineNo">196</span> private static void printUsage(final String errorMsg) {<a name="line.196"></a> +<span class="sourceLineNo">197</span> if (errorMsg != null && errorMsg.length() > 0) {<a name="line.197"></a> +<span class="sourceLineNo">198</span> System.err.println("ERROR: " + errorMsg);<a name="line.198"></a> +<span class="sourceLineNo">199</span> }<a name="line.199"></a> +<span class="sourceLineNo">200</span> System.err.println("Usage: CopyTable [general options] [--starttime=X] [--endtime=Y] " +<a name="line.200"></a> +<span class="sourceLineNo">201</span> "[--new.name=NEW] [--peer.adr=ADR] <tablename | snapshotName>");<a name="line.201"></a> +<span class="sourceLineNo">202</span> System.err.println();<a name="line.202"></a> +<span class="sourceLineNo">203</span> System.err.println("Options:");<a name="line.203"></a> +<span class="sourceLineNo">204</span> System.err.println(" rs.class hbase.regionserver.class of the peer cluster");<a name="line.204"></a> +<span class="sourceLineNo">205</span> System.err.println(" specify if different from current cluster");<a name="line.205"></a> +<span class="sourceLineNo">206</span> System.err.println(" rs.impl hbase.regionserver.impl of the peer cluster");<a name="line.206"></a> +<span class="sourceLineNo">207</span> System.err.println(" startrow the start row");<a name="line.207"></a> +<span class="sourceLineNo">208</span> System.err.println(" stoprow the stop row");<a name="line.208"></a> +<span class="sourceLineNo">209</span> System.err.println(" starttime beginning of the time range (unixtime in millis)");<a name="line.209"></a> +<span class="sourceLineNo">210</span> System.err.println(" without endtime means from starttime to forever");<a name="line.210"></a> +<span class="sourceLineNo">211</span> System.err.println(" endtime end of the time range. Ignored if no starttime specified.");<a name="line.211"></a> +<span class="sourceLineNo">212</span> System.err.println(" versions number of cell versions to copy");<a name="line.212"></a> +<span class="sourceLineNo">213</span> System.err.println(" new.name new table's name");<a name="line.213"></a> +<span class="sourceLineNo">214</span> System.err.println(" peer.adr Address of the peer cluster given in the format");<a name="line.214"></a> +<span class="sourceLineNo">215</span> System.err.println(" hbase.zookeeper.quorum:hbase.zookeeper.client"<a name="line.215"></a> +<span class="sourceLineNo">216</span> + ".port:zookeeper.znode.parent");<a name="line.216"></a> +<span class="sourceLineNo">217</span> System.err.println(" families comma-separated list of families to copy");<a name="line.217"></a> +<span class="sourceLineNo">218</span> System.err.println(" To copy from cf1 to cf2, give sourceCfName:destCfName. ");<a name="line.218"></a> +<span class="sourceLineNo">219</span> System.err.println(" To keep the same name, just give \"cfName\"");<a name="line.219"></a> +<span class="sourceLineNo">220</span> System.err.println(" all.cells also copy delete markers and deleted cells");<a name="line.220"></a> +<span class="sourceLineNo">221</span> System.err.println(" bulkload Write input into HFiles and bulk load to the destination "<a name="line.221"></a> +<span class="sourceLineNo">222</span> + "table");<a name="line.222"></a> +<span class="sourceLineNo">223</span> System.err.println(" snapshot Copy the data from snapshot to destination table.");<a name="line.223"></a> +<span class="sourceLineNo">224</span> System.err.println();<a name="line.224"></a> +<span class="sourceLineNo">225</span> System.err.println("Args:");<a name="line.225"></a> +<span class="sourceLineNo">226</span> System.err.println(" tablename Name of the table to copy");<a name="line.226"></a> +<span class="sourceLineNo">227</span> System.err.println();<a name="line.227"></a> +<span class="sourceLineNo">228</span> System.err.println("Examples:");<a name="line.228"></a> +<span class="sourceLineNo">229</span> System.err.println(" To copy 'TestTable' to a cluster that uses replication for a 1 hour window:");<a name="line.229"></a> +<span class="sourceLineNo">230</span> System.err.println(" $ hbase " +<a name="line.230"></a> +<span class="sourceLineNo">231</span> "org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1265875194289 --endtime=1265878794289 " +<a name="line.231"></a> +<span class="sourceLineNo">232</span> "--peer.adr=server1,server2,server3:2181:/hbase --families=myOldCf:myNewCf,cf2,cf3 TestTable ");<a name="line.232"></a> +<span class="sourceLineNo">233</span> System.err.println(" To copy data from 'sourceTableSnapshot' to 'destTable': ");<a name="line.233"></a> +<span class="sourceLineNo">234</span> System.err.println(" $ hbase org.apache.hadoop.hbase.mapreduce.CopyTable "<a name="line.234"></a> +<span class="sourceLineNo">235</span> + "--snapshot --new.name=destTable sourceTableSnapshot");<a name="line.235"></a> +<span class="sourceLineNo">236</span> System.err.println(" To copy data from 'sourceTableSnapshot' and bulk load to 'destTable': ");<a name="line.236"></a> +<span class="sourceLineNo">237</span> System.err.println(" $ hbase org.apache.hadoop.hbase.mapreduce.CopyTable "<a name="line.237"></a> +<span class="sourceLineNo">238</span> + "--new.name=destTable --snapshot --bulkload sourceTableSnapshot");<a name="line.238"></a> +<span class="sourceLineNo">239</span> System.err.println("For performance consider the following general option:\n"<a name="line.239"></a> +<span class="sourceLineNo">240</span> + " It is recommended that you set the following to >=100. A higher value uses more memory but\n"<a name="line.240"></a> +<span class="sourceLineNo">241</span> + " decreases the round trip time to the server and may increase performance.\n"<a name="line.241"></a> +<span class="sourceLineNo">242</span> + " -Dhbase.client.scanner.caching=100\n"<a name="line.242"></a> +<span class="sourceLineNo">243</span> + " The following should always be set to false, to prevent writing data twice, which may produce \n"<a name="line.243"></a> +<span class="sourceLineNo">244</span> + " inaccurate results.\n"<a name="line.244"></a> +<span class="sourceLineNo">245</span> + " -Dmapreduce.map.speculative=false");<a name="line.245"></a> +<span class="sourceLineNo">246</span> }<a name="line.246"></a> +<span class="sourceLineNo">247</span><a name="line.247"></a> +<span class="sourceLineNo">248</span> private boolean doCommandLine(final String[] args) {<a name="line.248"></a> +<span class="sourceLineNo">249</span> if (args.length < 1) {<a name="line.249"></a> +<span class="sourceLineNo">250</span> printUsage(null);<a name="line.250"></a> +<span class="sourceLineNo">251</span> return false;<a name="line.251"></a> +<span class="sourceLineNo">252</span> }<a name="line.252"></a> +<span class="sourceLineNo">253</span> try {<a name="line.253"></a> +<span class="sourceLineNo">254</span> for (int i = 0; i < args.length; i++) {<a name="line.254"></a> +<span class="sourceLineNo">255</span> String cmd = args[i];<a name="line.255"></a> +<span class="sourceLineNo">256</span> if (cmd.equals("-h") || cmd.startsWith("--h")) {<a name="line.256"></a> +<span class="sourceLineNo">257</span> printUsage(null);<a name="line.257"></a> +<span class="sourceLineNo">258</span> return false;<a name="line.258"></a> +<span class="sourceLineNo">259</span> }<a name="line.259"></a> +<span class="sourceLineNo">260</span><a name="line.260"></a> +<span class="sourceLineNo">261</span> final String startRowArgKey = "--startrow=";<a name="line.261"></a> +<span class="sourceLineNo">262</span> if (cmd.startsWith(startRowArgKey)) {<a name="line.262"></a> +<span class="sourceLineNo">263</span> startRow = cmd.substring(startRowArgKey.length());<a name="line.263"></a> +<span class="sourceLineNo">264</span> continue;<a name="line.264"></a> +<span class="sourceLineNo">265</span> }<a name="line.265"></a> +<span class="sourceLineNo">266</span><a name="line.266"></a> +<span class="sourceLineNo">267</span> final String stopRowArgKey = "--stoprow=";<a name="line.267"></a> +<span class="sourceLineNo">268</span> if (cmd.startsWith(stopRowArgKey)) {<a name="line.268"></a> +<span class="sourceLineNo">269</span> stopRow = cmd.substring(stopRowArgKey.length());<a name="line.269"></a> +<span class="sourceLineNo">270</span> continue;<a name="line.270"></a> +<span class="sourceLineNo">271</span> }<a name="line.271"></a> +<span class="sourceLineNo">272</span><a name="line.272"></a> +<span class="sourceLineNo">273</span> final String startTimeArgKey = "--starttime=";<a name="line.273"></a> +<span class="sourceLineNo">274</span> if (cmd.startsWith(startTimeArgKey)) {<a name="line.274"></a> +<span class="sourceLineNo">275</span> startTime = Long.parseLong(cmd.substring(startTimeArgKey.length()));<a name="line.275"></a> +<span class="sourceLineNo">276</span> continue;<a name="line.276"></a> +<span class="sourceLineNo">277</span> }<a name="line.277"></a> +<span class="sourceLineNo">278</span><a name="line.278"></a> +<span class="sourceLineNo">279</span> final String endTimeArgKey = "--endtime=";<a name="line.279"></a> +<span class="sourceLineNo">280</span> if (cmd.startsWith(endTimeArgKey)) {<a name="line.280"></a> +<span class="sourceLineNo">281</span> endTime = Long.parseLong(cmd.substring(endTimeArgKey.length()));<a name="line.281"></a> +<span class="sourceLineNo">282</span> continue;<a name="line.282"></a> +<span class="sourceLineNo">283</span> }<a name="line.283"></a> +<span class="sourceLineNo">284</span><a name="line.284"></a> +<span class="sourceLineNo">285</span> final String batchArgKey = "--batch=";<a name="line.285"></a> +<span class="sourceLineNo">286</span> if (cmd.startsWith(batchArgKey)) {<a name="line.286"></a> +<span class="sourceLineNo">287</span> batch = Integer.parseInt(cmd.substring(batchArgKey.length()));<a name="line.287"></a> +<span class="sourceLineNo">288</span> continue;<a name="line.288"></a> +<span class="sourceLineNo">289</span> }<a name="line.289"></a> +<span class="sourceLineNo">290</span><a name="line.290"></a> +<span class="sourceLineNo">291</span> final String cacheRowArgKey = "--cacheRow=";<a name="line.291"></a> +<span class="sourceLineNo">292</span> if (cmd.startsWith(cacheRowArgKey)) {<a name="line.292"></a> +<span class="sourceLineNo">293</span> cacheRow = Integer.parseInt(cmd.substring(cacheRowArgKey.length()));<a name="line.293"></a> +<span class="sourceLineNo">294</span> continue;<a name="line.294"></a> +<span class="sourceLineNo">295</span> }<a name="line.295"></a> +<span class="sourceLineNo">296</span><a name="line.296"></a> +<span class="sourceLineNo">297</span> final String versionsArgKey = "--versions=";<a name="line.297"></a> +<span class="sourceLineNo">298</span> if (cmd.startsWith(versionsArgKey)) {<a name="line.298"></a> +<span class="sourceLineNo">299</span> versions = Integer.parseInt(cmd.substring(versionsArgKey.length()));<a name="line.299"></a> +<span class="sourceLineNo">300</span> continue;<a name="line.300"></a> +<span class="sourceLineNo">301</span> }<a name="line.301"></a> +<span class="sourceLineNo">302</span><a name="line.302"></a> +<span class="sourceLineNo">303</span> final String newNameArgKey = "--new.name=";<a name="line.303"></a> +<span class="sourceLineNo">304</span> if (cmd.startsWith(newNameArgKey)) {<a name="line.304"></a> +<span class="sourceLineNo">305</span> dstTableName = cmd.substring(newNameArgKey.length());<a name="line.305"></a> +<span class="sourceLineNo">306</span> continue;<a name="line.306"></a> +<span class="sourceLineNo">307</span> }<a name="line.307"></a> +<span class="sourceLineNo">308</span><a name="line.308"></a> +<span class="sourceLineNo">309</span> final String peerAdrArgKey = "--peer.adr=";<a name="line.309"></a> +<span class="sourceLineNo">310</span> if (cmd.startsWith(peerAdrArgKey)) {<a name="line.310"></a> +<span class="sourceLineNo">311</span> peerAddress = cmd.substring(peerAdrArgKey.length());<a name="line.311"></a> +<span class="sourceLineNo">312</span> continue;<a name="line.312"></a> +<span class="sourceLineNo">313</span> }<a name="line.313"></a> +<span class="sourceLineNo">314</span><a name="line.314"></a> +<span class="sourceLineNo">315</span> final String familiesArgKey = "--families=";<a name="line.315"></a> +<span class="sourceLineNo">316</span> if (cmd.startsWith(familiesArgKey)) {<a name="line.316"></a> +<span class="sourceLineNo">317</span> families = cmd.substring(familiesArgKey.length());<a name="line.317"></a> +<span class="sourceLineNo">318</span> continue;<a name="line.318"></a> +<span class="sourceLineNo">319</span> }<a name="line.319"></a> +<span class="sourceLineNo">320</span><a name="line.320"></a> +<span class="sourceLineNo">321</span> if (cmd.startsWith("--all.cells")) {<a name="line.321"></a> +<span class="sourceLineNo">322</span> allCells = true;<a name="line.322"></a> +<span class="sourceLineNo">323</span> continue;<a name="line.323"></a> +<span class="sourceLineNo">324</span> }<a name="line.324"></a> +<span class="sourceLineNo">325</span><a name="line.325"></a> +<span class="sourceLineNo">326</span> if (cmd.startsWith("--bulkload")) {<a name="line.326"></a> +<span class="sourceLineNo">327</span> bulkload = true;<a name="line.327"></a> +<span class="sourceLineNo">328</span> continue;<a name="line.328"></a> +<span class="sourceLineNo">329</span> }<a name="line.329"></a> +<span class="sourceLineNo">330</span><a name="line.330"></a> +<span class="sourceLineNo">331</span> if (cmd.startsWith("--shuffle")) {<a name="line.331"></a> +<span class="sourceLineNo">332</span> shuffle = true;<a name="line.332"></a> +<span class="sourceLineNo">333</span> continue;<a name="line.333"></a> +<span class="sourceLineNo">334</span> }<a name="line.334"></a> +<span class="sourceLineNo">335</span><a name="line.335"></a> +<span class="sourceLineNo">336</span> if(cmd.startsWith("--snapshot")){<a name="line.336"></a> +<span class="sourceLineNo">337</span> readingSnapshot = true;<a name="line.337"></a> +<span class="sourceLineNo">338</span> continue;<a name="line.338"></a> +<span class="sourceLineNo">339</span> }<a name="line.339"></a> +<span class="sourceLineNo">340</span><a name="line.340"></a> +<span class="sourceLineNo">341</span> if (i == args.length - 1) {<a name="line.341"></a> +<span class="sourceLineNo">342</span> if (readingSnapshot) {<a name="line.342"></a> +<span class="sourceLineNo">343</span> snapshot = cmd;<a name="line.343"></a> +<span class="sourceLineNo">344</span> } else {<a name="line.344"></a> +<span class="sourceLineNo">345</span> tableName = cmd;<a name="line.345"></a> +<span class="sourceLineNo">346</span> }<a name="line.346"></a> +<span class="sourceLineNo">347</span> } else {<a name="line.347"></a> +<span class="sourceLineNo">348</span> printUsage("Invalid argument '" + cmd + "'");<a name="line.348"></a> +<span class="sourceLineNo">349</span> return false;<a name="line.349"></a> +<span class="sourceLineNo">350</span> }<a name="line.350"></a> +<span class="sourceLineNo">351</span> }<a name="line.351"></a> +<span class="sourceLineNo">352</span> if (dstTableName == null && peerAddress == null) {<a name="line.352"></a> +<span class="sourceLineNo">353</span> printUsage("At least a new table name or a peer address must be specified");<a name="line.353"></a> +<span class="sourceLineNo">354</span> return false;<a name="line.354"></a> +<span class="sourceLineNo">355</span> }<a name="line.355"></a> +<span class="sourceLineNo">356</span> if ((endTime != 0) && (startTime > endTime)) {<a name="line.356"></a> +<span class="sourceLineNo">357</span> printUsage("Invalid time range filter: starttime=" + startTime + " > endtime=" + endTime);<a name="line.357"></a> +<span class="sourceLineNo">358</span> return false;<a name="line.358"></a> +<span class="sourceLineNo">359</span> }<a name="line.359"></a> <span class="sourceLineNo">360</span><a name="line.360"></a> -<span class="sourceLineNo">361</span> @Override<a name="line.361"></a> -<span class="sourceLineNo">362</span> public int run(String[] args) throws Exception {<a name="line.362"></a> -<span class="sourceLineNo">363</span> Job job = createSubmittableJob(args);<a name="line.363"></a> -<span class="sourceLineNo">364</span> if (job == null) return 1;<a name="line.364"></a> -<span class="sourceLineNo">365</span> if (!job.waitForCompletion(true)) {<a name="line.365"></a> -<span class="sourceLineNo">366</span> LOG.info("Map-reduce job failed!");<a name="line.366"></a> -<span class="sourceLineNo">367</span> if (bulkload) {<a name="line.367"></a> -<span class="sourceLineNo">368</span> LOG.info("Files are not bulkloaded!");<a name="line.368"></a> +<span class="sourceLineNo">361</span> if (bulkload && peerAddress != null) {<a name="line.361"></a> +<span class="sourceLineNo">362</span> printUsage("Remote bulkload is not supported!");<a name="line.362"></a> +<span class="sourceLineNo">363</span> return false;<a name="line.363"></a> +<span class="sourceLineNo">364</span> }<a name="line.364"></a> +<span class="sourceLineNo">365</span><a name="line.365"></a> +<span class="sourceLineNo">366</span> if (readingSnapshot && peerAddress != null) {<a name="line.366"></a> +<span class="sourceLineNo">367</span> printUsage("Loading data from snapshot to remote peer cluster is not supported.");<a name="line.367"></a> +<span class="sourceLineNo">368</span> return false;<a name="line.368"></a> <span class="sourceLineNo">369</span> }<a name="line.369"></a> -<span class="sourceLineNo">370</span> return 1;<a name="line.370"></a> -<span class="sourceLineNo">371</span> }<a name="line.371"></a> -<span class="sourceLineNo">372</span> int code = 0;<a name="line.372"></a> -<span class="sourceLineNo">373</span> if (bulkload) {<a name="line.373"></a> -<span class="sourceLineNo">374</span> code = new LoadIncrementalHFiles(this.getConf())<a name="line.374"></a> -<span class="sourceLineNo">375</span> .run(new String[] { this.bulkloadDir.toString(), this.dstTableName });<a name="line.375"></a> -<span class="sourceLineNo">376</span> if (code == 0) {<a name="line.376"></a> -<span class="sourceLineNo">377</span> // bulkloadDir is deleted only LoadIncrementalHFiles was successful so that one can rerun<a name="line.377"></a> -<span class="sourceLineNo">378</span> // LoadIncrementalHFiles.<a name="line.378"></a> -<span class="sourceLineNo">379</span> FileSystem fs = FSUtils.getCurrentFileSystem(getConf());<a name="line.379"></a> -<span class="sourceLineNo">380</span> if (!fs.delete(this.bulkloadDir, true)) {<a name="line.380"></a> -<span class="sourceLineNo">381</span> LOG.error("Deleting folder " + bulkloadDir + " failed!");<a name="line.381"></a> -<span class="sourceLineNo">382</span> code = 1;<a name="line.382"></a> -<span class="sourceLineNo">383</span> }<a name="line.383"></a> -<span class="sourceLineNo">384</span> }<a name="line.384"></a> -<span class="sourceLineNo">385</span> }<a name="line.385"></a> -<span class="sourceLineNo">386</span> return code;<a name="line.386"></a> -<span class="sourceLineNo">387</span> }<a name="line.387"></a> -<span class="sourceLineNo">388</span>}<a name="line.388"></a> +<span class="sourceLineNo">370</span><a name="line.370"></a> +<span class="sourceLineNo">371</span> if (readingSnapshot && dstTableName == null) {<a name="line.371"></a> +<span class="sourceLineNo">372</span> printUsage("The --new.name=<table> for destination table should be "<a name="line.372"></a> +<span class="sourceLineNo">373</span> + "provided when copying data from snapshot .");<a name="line.373"></a> +<span class="sourceLineNo">374</span> return false;<a name="line.374"></a> +<span class="sourceLineNo">375</span> }<a name="line.375"></a> +<span class="sourceLineNo">376</span><a name="line.376"></a> +<span class="sourceLineNo">377</span> if (readingSnapshot && snapshot == null) {<a name="line.377"></a> +<span class="sourceLineNo">378</span> printUsage("Snapshot shouldn't be null when --snapshot is enabled.");<a name="line.378"></a> +<span class="sourceLineNo">379</span> return false;<a name="line.379"></a> +<span class="sourceLineNo">380</span> }<a name="line.380"></a> +<span class="sourceLineNo">381</span><a name="line.381"></a> +<span class="sourceLineNo">382</span> // set dstTableName if necessary<a name="line.382"></a> +<span class="sourceLineNo">383</span> if (dstTableName == null) {<a name="line.383"></a> +<span class="sourceLineNo">384</span> dstTableName = tableName;<a name="line.384"></a> +<span class="sourceLineNo">385</span> }<a name="line.385"></a> +<span class="sourceLineNo">386</span> } catch (Exception e) {<a name="line.386"></a> +<span class="sourceLineNo">387</span> LOG.error("Failed to parse commandLine arguments", e);<a name="line.387"></a> +<span class="sourceLineNo">388</span> printUsage("Can't start because " + e.getMessage());<a name="line.388"></a> +<span class="sourceLineNo">389</span> return false;<a name="line.389"></a> +<span class="sourceLineNo">390</span> }<a name="line.390"></a> +<span class="sourceLineNo">391</span> return true;<a name="line.391"></a> +<span class="sourceLineNo">392</span> }<a name="line.392"></a> +<span class="sourceLineNo">393</span><a name="line.393"></a> +<span class="sourceLineNo">394</span> /**<a name="line.394"></a> +<span class="sourceLineNo">395</span> * Main entry point.<a name="line.395"></a> +<span class="sourceLineNo">396</span> *<a name="line.396"></a> +<span class="sourceLineNo">397</span> * @param args The command line parameters.<a name="line.397"></a> +<span class="sourceLineNo">398</span> * @throws Exception When running the job fails.<a name="line.398"></a> +<span class="sourceLineNo">399</span> */<a name="line.399"></a> +<span class="sourceLineNo">400</span> public static void main(String[] args) throws Exception {<a name="line.400"></a> +<span class="sourceLineNo">401</span> int ret = ToolRunner.run(HBaseConfiguration.create(), new CopyTable(), args);<a name="line.401"></a> +<span class="sourceLineNo">402</span> System.exit(ret);<a name="line.402"></a> +<span class="sourceLineNo">403</span> }<a name="line.403"></a> +<span class="sourceLineNo">404</span><a name="line.404"></a> +<span class="sourceLineNo">405</span> @Override<a name="line.405"></a> +<span class="sourceLineNo">406</span> public int run(String[] args) throws Exception {<a name="line.406"></a> +<span class="sourceLineNo">407</span> Job job = createSubmittableJob(args);<a name="line.407"></a> +<span class="sourceLineNo">408</span> if (job == null) return 1;<a name="line.408"></a> +<span class="sourceLineNo">409</span> if (!job.waitForCompletion(true)) {<a name="line.409"></a> +<span class="sourceLineNo">410</span> LOG.info("Map-reduce job failed!");<a name="line.410"></a> +<span class="sourceLineNo">411</span> if (bulkload) {<a name="line.411"></a> +<span class="sourceLineNo">412</span> LOG.info("Files are not bulkloaded!");<a name="line.412"></a> +<span class="sourceLineNo">413</span> }<a name="line.413"></a> +<span class="sourceLineNo">414</span> return 1;<a name="line.414"></a> +<span class="sourceLineNo">415</span> }<a name="line.415"></a> +<span class="sourceLineNo">416</span> int code = 0;<a name="line.416"></a> +<span class="sourceLineNo">417</span> if (bulkload) {<a name="line.417"></a> +<span class="sourceLineNo">418</span> LOG.info("Trying to bulk load data to destination table: " + dstTableName);<a name="line.418"></a> +<span class="sourceLineNo">419</span> LOG.info("command: ./bin/hbase org.apache.hadoop.hbase.tool.LoadIncrementalHFiles {} {}",<a name="line.419"></a> +<span class="sourceLineNo">420</span> this.bulkloadDir.toString(), this.dstTableName);<a name="line.420"></a> +<span class="sourceLineNo">421</span> code = new LoadIncrementalHFiles(this.getConf())<a name="line.421"></a> +<span class="sourceLineNo">422</span> .run(new String[] { this.bulkloadDir.toString(), this.dstTableName });<a name="line.422"></a> +<span class="sourceLineNo">423</span> if (code == 0) {<a name="line.423"></a> +<span class="sourceLineNo">424</span> // bulkloadDir is deleted only LoadIncrementalHFiles was successful so that one can rerun<a name="line.424"></a> +<span class="sourceLineNo">425</span> // LoadIncrementalHFiles.<a name="line.425"></a> +<span class="sourceLineNo">426</span> FileSystem fs = FSUtils.getCurrentFileSystem(getConf());<a name="line.426"></a> +<span class="sourceLineNo">427</span> if (!fs.delete(this.bulkloadDir, true)) {<a name="line.427"></a> +<span class="sourceLineNo">428</span> LOG.error("Deleting folder " + bulkloadDir + " failed!");<a name="line.428"></a> +<span class="sourceLineNo">429</span> code = 1;<a name="line.429"></a> +<span class="sourceLineNo">430</span> }<a name="line.430"></a> +<span class="sourceLineNo">431</span> }<a name="line.431"></a> +<span class="sourceLineNo">432</span> }<a name="line.432"></a> +<span class="sourceLineNo">433</span> return code;<a name="line.433"></a> +<span class="sourceLineNo">434</span> }<a name="line.434"></a> +<span class="sourceLineNo">435</span>}<a name="line.435"></a> http://git-wip-us.apache.org/repos/asf/hbase-site/blob/09ea0d5f/book.html ---------------------------------------------------------------------- diff --git a/book.html b/book.html index af029d9..7663020 100644 --- a/book.html +++ b/book.html @@ -17825,19 +17825,12 @@ Using bulk load will use less CPU and network resources than simply using the HB </div> </div> <div class="sect2"> -<h3 id="arch.bulk.load.limitations"><a class="anchor" href="#arch.bulk.load.limitations"></a>73.2. Bulk Load Limitations</h3> -<div class="paragraph"> -<p>As bulk loading bypasses the write path, the WAL doesn’t get written to as part of the process. -Replication works by reading the WAL files so it won’t see the bulk loaded data â and the same goes for the edits that use <code>Put.setDurability(SKIP_WAL)</code>. One way to handle that is to ship the raw files or the HFiles to the other cluster and do the other processing there.</p> -</div> -</div> -<div class="sect2"> -<h3 id="arch.bulk.load.arch"><a class="anchor" href="#arch.bulk.load.arch"></a>73.3. Bulk Load Architecture</h3> +<h3 id="arch.bulk.load.arch"><a class="anchor" href="#arch.bulk.load.arch"></a>73.2. Bulk Load Architecture</h3> <div class="paragraph"> <p>The HBase bulk load process consists of two main steps.</p> </div> <div class="sect3"> -<h4 id="arch.bulk.load.prep"><a class="anchor" href="#arch.bulk.load.prep"></a>73.3.1. Preparing data via a MapReduce job</h4> +<h4 id="arch.bulk.load.prep"><a class="anchor" href="#arch.bulk.load.prep"></a>73.2.1. Preparing data via a MapReduce job</h4> <div class="paragraph"> <p>The first step of a bulk load is to generate HBase data files (StoreFiles) from a MapReduce job using <code>HFileOutputFormat2</code>. This output format writes out data in HBase’s internal storage format so that they can be later loaded very efficiently into the cluster.</p> @@ -17851,7 +17844,7 @@ In order to do this, jobs whose output will be bulk loaded into HBase use Hadoop </div> </div> <div class="sect3"> -<h4 id="arch.bulk.load.complete"><a class="anchor" href="#arch.bulk.load.complete"></a>73.3.2. Completing the data load</h4> +<h4 id="arch.bulk.load.complete"><a class="anchor" href="#arch.bulk.load.complete"></a>73.2.2. Completing the data load</h4> <div class="paragraph"> <p>After a data import has been prepared, either by using the <code>importtsv</code> tool with the “importtsv.bulk.output” option or by some other MapReduce job using the <code>HFileOutputFormat</code>, the <code>completebulkload</code> tool is used to import the data into the running cluster. This command line tool iterates through the prepared data files, and for each one determines the region the file belongs to. @@ -17884,7 +17877,7 @@ If the target table does not already exist in HBase, this tool will create the t </div> </div> <div class="sect2"> -<h3 id="arch.bulk.load.also"><a class="anchor" href="#arch.bulk.load.also"></a>73.4. See Also</h3> +<h3 id="arch.bulk.load.also"><a class="anchor" href="#arch.bulk.load.also"></a>73.3. See Also</h3> <div class="paragraph"> <p>For more information about the referenced utilities, see <a href="#importtsv">ImportTsv</a> and <a href="#completebulkload">CompleteBulkLoad</a>.</p> </div> @@ -17893,7 +17886,7 @@ If the target table does not already exist in HBase, this tool will create the t </div> </div> <div class="sect2"> -<h3 id="arch.bulk.load.adv"><a class="anchor" href="#arch.bulk.load.adv"></a>73.5. Advanced Usage</h3> +<h3 id="arch.bulk.load.adv"><a class="anchor" href="#arch.bulk.load.adv"></a>73.4. Advanced Usage</h3> <div class="paragraph"> <p>Although the <code>importtsv</code> tool is useful in many cases, advanced users may want to generate data programmatically, or import data from other formats. To get started doing so, dig into <code>ImportTsv.java</code> and check the JavaDoc for HFileOutputFormat.</p> @@ -17903,6 +17896,53 @@ To get started doing so, dig into <code>ImportTsv.java</code> and check the Java See the <code>LoadIncrementalHFiles</code> class for more information.</p> </div> </div> +<div class="sect2"> +<h3 id="arch.bulk.load.replication"><a class="anchor" href="#arch.bulk.load.replication"></a>73.5. Bulk Loading Replication</h3> +<div class="paragraph"> +<p>HBASE-13153 adds replication support for bulk loaded HFiles, available since HBase 1.3/2.0. This feature is enabled by setting <code>hbase.replication.bulkload.enabled</code> to <code>true</code> (default is <code>false</code>). +You also need to copy the source cluster configuration files to the destination cluster.</p> +</div> +<div class="paragraph"> +<p>Additional configurations are required too:</p> +</div> +<div class="olist arabic"> +<ol class="arabic"> +<li> +<p><code>hbase.replication.source.fs.conf.provider</code></p> +<div class="paragraph"> +<p>This defines the class which loads the source cluster file system client configuration in the destination cluster. This should be configured for all the RS in the destination cluster. Default is <code>org.apache.hadoop.hbase.replication.regionserver.DefaultSourceFSConfigurationProvider</code>.</p> +</div> +</li> +<li> +<p><code>hbase.replication.conf.dir</code></p> +<div class="paragraph"> +<p>This represents the base directory where the file system client configurations of the source cluster are copied to the destination cluster. This should be configured for all the RS in the destination cluster. Default is <code>$HBASE_CONF_DIR</code>.</p> +</div> +</li> +<li> +<p><code>hbase.replication.cluster.id</code></p> +<div class="paragraph"> +<p>This configuration is required in the cluster where replication for bulk loaded data is enabled. A source cluster is uniquely identified by the destination cluster using this id. This should be configured for all the RS in the source cluster configuration file for all the RS.</p> +</div> +</li> +</ol> +</div> +<div class="paragraph"> +<p>For example: If source cluster FS client configurations are copied to the destination cluster under directory <code>/home/user/dc1/</code>, then <code>hbase.replication.cluster.id</code> should be configured as <code>dc1</code> and <code>hbase.replication.conf.dir</code> as <code>/home/user</code>.</p> +</div> +<div class="admonitionblock note"> +<table> +<tr> +<td class="icon"> +<i class="fa icon-note" title="Note"></i> +</td> +<td class="content"> +<code>DefaultSourceFSConfigurationProvider</code> supports only <code>xml</code> type files. It loads source cluster FS client configuration only once, so if source cluster FS client configuration files are updated, every peer(s) cluster RS must be restarted to reload the configuration. +</td> +</tr> +</table> +</div> +</div> </div> </div> <div class="sect1"> @@ -41371,7 +41411,7 @@ org/apache/hadoop/hbase/security/access/AccessControlClient.revoke:(Lorg/apache/ <div id="footer"> <div id="footer-text"> Version 3.0.0-SNAPSHOT<br> -Last updated 2018-12-26 14:32:40 UTC +Last updated 2018-12-27 14:32:59 UTC </div> </div> </body> http://git-wip-us.apache.org/repos/asf/hbase-site/blob/09ea0d5f/bulk-loads.html ---------------------------------------------------------------------- diff --git a/bulk-loads.html b/bulk-loads.html index 2a9316d..9dd66d8 100644 --- a/bulk-loads.html +++ b/bulk-loads.html @@ -7,7 +7,7 @@ <head> <meta charset="UTF-8" /> <meta name="viewport" content="width=device-width, initial-scale=1.0" /> - <meta name="Date-Revision-yyyymmdd" content="20181226" /> + <meta name="Date-Revision-yyyymmdd" content="20181227" /> <meta http-equiv="Content-Language" content="en" /> <title>Apache HBase – Bulk Loads in Apache HBase (TM) @@ -316,7 +316,7 @@ under the License. --> <a href="https://www.apache.org/";>The Apache Software Foundation</a>. All rights reserved. - <li id="publishDate" class="pull-right">Last Published: 2018-12-26</li> + <li id="publishDate" class="pull-right">Last Published: 2018-12-27</li> </p> </div>
