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.TreeMap;
23  
24  import org.apache.hadoop.conf.Configuration;
25  import org.apache.hadoop.conf.Configured;
26  import org.apache.hadoop.hbase.HBaseConfiguration;
27  import org.apache.hadoop.hbase.client.Put;
28  import org.apache.hadoop.hbase.client.Result;
29  import org.apache.hadoop.hbase.client.Scan;
30  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
31  import org.apache.hadoop.hbase.util.Bytes;
32  import org.apache.hadoop.mapreduce.Job;
33  import org.apache.hadoop.mapreduce.Mapper;
34  import org.apache.hadoop.util.Tool;
35  import org.apache.hadoop.util.ToolRunner;
36  
37  /**
38   * Example map/reduce job to construct index tables that can be used to quickly
39   * find a row based on the value of a column. It demonstrates:
40   * <ul>
41   * <li>Using TableInputFormat and TableMapReduceUtil to use an HTable as input
42   * to a map/reduce job.</li>
43   * <li>Passing values from main method to children via the configuration.</li>
44   * <li>Using MultiTableOutputFormat to output to multiple tables from a
45   * map/reduce job.</li>
46   * <li>A real use case of building a secondary index over a table.</li>
47   * </ul>
48   *
49   * <h3>Usage</h3>
50   *
51   * <p>
52   * Modify ${HADOOP_HOME}/conf/hadoop-env.sh to include the hbase jar, the
53   * zookeeper jar (can be found in lib/ directory under HBase root, the examples output directory,
54   * and the hbase conf directory in HADOOP_CLASSPATH, and then run
55   * <tt><strong>bin/hadoop org.apache.hadoop.hbase.mapreduce.IndexBuilder TABLE_NAME COLUMN_FAMILY ATTR [ATTR ...]</strong></tt>
56   * </p>
57   *
58   * <p>
59   * To run with the sample data provided in index-builder-setup.rb, use the
60   * arguments <strong><tt>people attributes name email phone</tt></strong>.
61   * </p>
62   *
63   * <p>
64   * This code was written against HBase 0.21 trunk.
65   * </p>
66   */
67  public class IndexBuilder extends Configured implements Tool {
68    /** the column family containing the indexed row key */
69    public static final byte[] INDEX_COLUMN = Bytes.toBytes("INDEX");
70    /** the qualifier containing the indexed row key */
71    public static final byte[] INDEX_QUALIFIER = Bytes.toBytes("ROW");
72  
73    /**
74     * Internal Mapper to be run by Hadoop.
75     */
76    public static class Map extends
77        Mapper<ImmutableBytesWritable, Result, ImmutableBytesWritable, Put> {
78      private byte[] family;
79      private TreeMap<byte[], ImmutableBytesWritable> indexes;
80  
81      @Override
82      protected void map(ImmutableBytesWritable rowKey, Result result, Context context)
83          throws IOException, InterruptedException {
84        for(java.util.Map.Entry<byte[], ImmutableBytesWritable> index : indexes.entrySet()) {
85          byte[] qualifier = index.getKey();
86          ImmutableBytesWritable tableName = index.getValue();
87          byte[] value = result.getValue(family, qualifier);
88          if (value != null) {
89            // original: row 123 attribute:phone 555-1212
90            // index: row 555-1212 INDEX:ROW 123
91            Put put = new Put(value);
92            put.add(INDEX_COLUMN, INDEX_QUALIFIER, rowKey.get());
93            context.write(tableName, put);
94          }
95        }
96      }
97  
98      @Override
99      protected void setup(Context context) throws IOException,
100         InterruptedException {
101       Configuration configuration = context.getConfiguration();
102       String tableName = configuration.get("index.tablename");
103       String[] fields = configuration.getStrings("index.fields");
104       String familyName = configuration.get("index.familyname");
105       family = Bytes.toBytes(familyName);
106       indexes = new TreeMap<byte[], ImmutableBytesWritable>(Bytes.BYTES_COMPARATOR);
107       for(String field : fields) {
108         // if the table is "people" and the field to index is "email", then the
109         // index table will be called "people-email"
110         indexes.put(Bytes.toBytes(field),
111             new ImmutableBytesWritable(Bytes.toBytes(tableName + "-" + field)));
112       }
113     }
114   }
115 
116   /**
117    * Job configuration.
118    */
119   public static Job configureJob(Configuration conf, String [] args)
120   throws IOException {
121     String tableName = args[0];
122     String columnFamily = args[1];
123     System.out.println("****" + tableName);
124     conf.set(TableInputFormat.SCAN, TableMapReduceUtil.convertScanToString(new Scan()));
125     conf.set(TableInputFormat.INPUT_TABLE, tableName);
126     conf.set("index.tablename", tableName);
127     conf.set("index.familyname", columnFamily);
128     String[] fields = new String[args.length - 2];
129     System.arraycopy(args, 2, fields, 0, fields.length);
130     conf.setStrings("index.fields", fields);
131     Job job = new Job(conf, tableName);
132     job.setJarByClass(IndexBuilder.class);
133     job.setMapperClass(Map.class);
134     job.setNumReduceTasks(0);
135     job.setInputFormatClass(TableInputFormat.class);
136     job.setOutputFormatClass(MultiTableOutputFormat.class);
137     return job;
138   }
139 
140   public int run(String[] args) throws Exception {
141     Configuration conf = HBaseConfiguration.create(getConf());
142     if(args.length < 3) {
143       System.err.println("Only " + args.length + " arguments supplied, required: 3");
144       System.err.println("Usage: IndexBuilder <TABLE_NAME> <COLUMN_FAMILY> <ATTR> [<ATTR> ...]");
145       System.exit(-1);
146     }
147     Job job = configureJob(conf, args);
148     return (job.waitForCompletion(true) ? 0 : 1);
149   }
150 
151   public static void main(String[] args) throws Exception {
152     int result = ToolRunner.run(new Configuration(), new IndexBuilder(), args);
153     System.exit(result);
154   }
155 }