View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.mapreduce;
20  
21  import java.io.IOException;
22  import java.util.Collections;
23  import java.util.List;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.conf.Configurable;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.hbase.KeyValue;
30  import org.apache.hadoop.hbase.TableName;
31  import org.apache.hadoop.hbase.classification.InterfaceAudience;
32  import org.apache.hadoop.hbase.classification.InterfaceStability;
33  import org.apache.hadoop.hbase.client.Connection;
34  import org.apache.hadoop.hbase.client.ConnectionFactory;
35  import org.apache.hadoop.hbase.client.RegionLocator;
36  import org.apache.hadoop.hbase.client.Scan;
37  import org.apache.hadoop.hbase.util.Bytes;
38  import org.apache.hadoop.mapreduce.InputSplit;
39  import org.apache.hadoop.mapreduce.JobContext;
40  import org.apache.hadoop.hbase.util.Pair;
41  import org.apache.hadoop.mapreduce.Job;
42  import org.apache.hadoop.util.StringUtils;
43  
44  /**
45   * Convert HBase tabular data into a format that is consumable by Map/Reduce.
46   */
47  @InterfaceAudience.Public
48  @InterfaceStability.Stable
49  public class TableInputFormat extends TableInputFormatBase
50  implements Configurable {
51  
52    private static final Log LOG = LogFactory.getLog(TableInputFormat.class);
53  
54    /** Job parameter that specifies the input table. */
55    public static final String INPUT_TABLE = "hbase.mapreduce.inputtable";
56    /**
57     * If specified, use start keys of this table to split.
58     * This is useful when you are preparing data for bulkload.
59     */
60    private static final String SPLIT_TABLE = "hbase.mapreduce.splittable";
61    /** Base-64 encoded scanner. All other SCAN_ confs are ignored if this is specified.
62     * See {@link TableMapReduceUtil#convertScanToString(Scan)} for more details.
63     */
64    public static final String SCAN = "hbase.mapreduce.scan";
65    /** Scan start row */
66    public static final String SCAN_ROW_START = "hbase.mapreduce.scan.row.start";
67    /** Scan stop row */
68    public static final String SCAN_ROW_STOP = "hbase.mapreduce.scan.row.stop";
69    /** Column Family to Scan */
70    public static final String SCAN_COLUMN_FAMILY = "hbase.mapreduce.scan.column.family";
71    /** Space delimited list of columns and column families to scan. */
72    public static final String SCAN_COLUMNS = "hbase.mapreduce.scan.columns";
73    /** The timestamp used to filter columns with a specific timestamp. */
74    public static final String SCAN_TIMESTAMP = "hbase.mapreduce.scan.timestamp";
75    /** The starting timestamp used to filter columns with a specific range of versions. */
76    public static final String SCAN_TIMERANGE_START = "hbase.mapreduce.scan.timerange.start";
77    /** The ending timestamp used to filter columns with a specific range of versions. */
78    public static final String SCAN_TIMERANGE_END = "hbase.mapreduce.scan.timerange.end";
79    /** The maximum number of version to return. */
80    public static final String SCAN_MAXVERSIONS = "hbase.mapreduce.scan.maxversions";
81    /** Set to false to disable server-side caching of blocks for this scan. */
82    public static final String SCAN_CACHEBLOCKS = "hbase.mapreduce.scan.cacheblocks";
83    /** The number of rows for caching that will be passed to scanners. */
84    public static final String SCAN_CACHEDROWS = "hbase.mapreduce.scan.cachedrows";
85    /** Set the maximum number of values to return for each call to next(). */
86    public static final String SCAN_BATCHSIZE = "hbase.mapreduce.scan.batchsize";
87    /** Specify if we have to shuffle the map tasks. */
88    public static final String SHUFFLE_MAPS = "hbase.mapreduce.inputtable.shufflemaps";
89  
90    /** The configuration. */
91    private Configuration conf = null;
92  
93    /**
94     * Returns the current configuration.
95     *
96     * @return The current configuration.
97     * @see org.apache.hadoop.conf.Configurable#getConf()
98     */
99    @Override
100   public Configuration getConf() {
101     return conf;
102   }
103 
104   /**
105    * Sets the configuration. This is used to set the details for the table to
106    * be scanned.
107    *
108    * @param configuration  The configuration to set.
109    * @see org.apache.hadoop.conf.Configurable#setConf(
110    *   org.apache.hadoop.conf.Configuration)
111    */
112   @Override
113   public void setConf(Configuration configuration) {
114     this.conf = configuration;
115     TableName tableName = TableName.valueOf(conf.get(INPUT_TABLE));
116     try {
117       // NOTE: This connection doesn't currently get closed explicit1ly.
118       initializeTable(ConnectionFactory.createConnection(new Configuration(conf)), tableName);
119     } catch (Exception e) {
120       LOG.error(StringUtils.stringifyException(e));
121     }
122 
123     Scan scan = null;
124 
125     if (conf.get(SCAN) != null) {
126       try {
127         scan = TableMapReduceUtil.convertStringToScan(conf.get(SCAN));
128       } catch (IOException e) {
129         LOG.error("An error occurred.", e);
130       }
131     } else {
132       try {
133         scan = new Scan();
134 
135         if (conf.get(SCAN_ROW_START) != null) {
136           scan.setStartRow(Bytes.toBytes(conf.get(SCAN_ROW_START)));
137         }
138 
139         if (conf.get(SCAN_ROW_STOP) != null) {
140           scan.setStopRow(Bytes.toBytes(conf.get(SCAN_ROW_STOP)));
141         }
142 
143         if (conf.get(SCAN_COLUMNS) != null) {
144           addColumns(scan, conf.get(SCAN_COLUMNS));
145         }
146 
147         if (conf.get(SCAN_COLUMN_FAMILY) != null) {
148           scan.addFamily(Bytes.toBytes(conf.get(SCAN_COLUMN_FAMILY)));
149         }
150 
151         if (conf.get(SCAN_TIMESTAMP) != null) {
152           scan.setTimeStamp(Long.parseLong(conf.get(SCAN_TIMESTAMP)));
153         }
154 
155         if (conf.get(SCAN_TIMERANGE_START) != null && conf.get(SCAN_TIMERANGE_END) != null) {
156           scan.setTimeRange(
157               Long.parseLong(conf.get(SCAN_TIMERANGE_START)),
158               Long.parseLong(conf.get(SCAN_TIMERANGE_END)));
159         }
160 
161         if (conf.get(SCAN_MAXVERSIONS) != null) {
162           scan.setMaxVersions(Integer.parseInt(conf.get(SCAN_MAXVERSIONS)));
163         }
164 
165         if (conf.get(SCAN_CACHEDROWS) != null) {
166           scan.setCaching(Integer.parseInt(conf.get(SCAN_CACHEDROWS)));
167         }
168 
169         if (conf.get(SCAN_BATCHSIZE) != null) {
170           scan.setBatch(Integer.parseInt(conf.get(SCAN_BATCHSIZE)));
171         }
172 
173         // false by default, full table scans generate too much BC churn
174         scan.setCacheBlocks((conf.getBoolean(SCAN_CACHEBLOCKS, false)));
175       } catch (Exception e) {
176           LOG.error(StringUtils.stringifyException(e));
177       }
178     }
179 
180     setScan(scan);
181   }
182 
183   /**
184    * Parses a combined family and qualifier and adds either both or just the
185    * family in case there is no qualifier. This assumes the older colon
186    * divided notation, e.g. "family:qualifier".
187    *
188    * @param scan The Scan to update.
189    * @param familyAndQualifier family and qualifier
190    * @throws IllegalArgumentException When familyAndQualifier is invalid.
191    */
192   private static void addColumn(Scan scan, byte[] familyAndQualifier) {
193     byte [][] fq = KeyValue.parseColumn(familyAndQualifier);
194     if (fq.length == 1) {
195       scan.addFamily(fq[0]);
196     } else if (fq.length == 2) {
197       scan.addColumn(fq[0], fq[1]);
198     } else {
199       throw new IllegalArgumentException("Invalid familyAndQualifier provided.");
200     }
201   }
202 
203   /**
204    * Adds an array of columns specified using old format, family:qualifier.
205    * <p>
206    * Overrides previous calls to {@link Scan#addColumn(byte[], byte[])}for any families in the
207    * input.
208    *
209    * @param scan The Scan to update.
210    * @param columns array of columns, formatted as <code>family:qualifier</code>
211    * @see Scan#addColumn(byte[], byte[])
212    */
213   public static void addColumns(Scan scan, byte [][] columns) {
214     for (byte[] column : columns) {
215       addColumn(scan, column);
216     }
217   }
218 
219   /**
220    * Calculates the splits that will serve as input for the map tasks. The
221    * number of splits matches the number of regions in a table. Splits are shuffled if
222    * required.
223    * @param context  The current job context.
224    * @return The list of input splits.
225    * @throws IOException When creating the list of splits fails.
226    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
227    *   org.apache.hadoop.mapreduce.JobContext)
228    */
229   @Override
230   public List<InputSplit> getSplits(JobContext context) throws IOException {
231     List<InputSplit> splits = super.getSplits(context);
232     if ((conf.get(SHUFFLE_MAPS) != null) && "true".equals(conf.get(SHUFFLE_MAPS).toLowerCase())) {
233       Collections.shuffle(splits);
234     }
235     return splits;
236   }
237 
238   /**
239    * Convenience method to parse a string representation of an array of column specifiers.
240    *
241    * @param scan The Scan to update.
242    * @param columns  The columns to parse.
243    */
244   private static void addColumns(Scan scan, String columns) {
245     String[] cols = columns.split(" ");
246     for (String col : cols) {
247       addColumn(scan, Bytes.toBytes(col));
248     }
249   }
250 
251   @Override
252   protected Pair<byte[][], byte[][]> getStartEndKeys() throws IOException {
253     if (conf.get(SPLIT_TABLE) != null) {
254       TableName splitTableName = TableName.valueOf(conf.get(SPLIT_TABLE));
255       try (Connection conn = ConnectionFactory.createConnection(getConf());
256           RegionLocator rl = conn.getRegionLocator(splitTableName)) {
257         return rl.getStartEndKeys();
258       }
259     }
260 
261     return super.getStartEndKeys();
262   }
263 
264   /**
265    * Sets split table in map-reduce job.
266    */
267   public static void configureSplitTable(Job job, TableName tableName) {
268     job.getConfiguration().set(SPLIT_TABLE, tableName.getNameAsString());
269   }
270 }