· 9 years ago · Oct 31, 2016, 04:54 PM
1
2// Standard classes
3import java.io.IOException;
4import java.util.Iterator;
5
6// HBase classes
7import org.apache.hadoop.hbase.HBaseConfiguration;
8import org.apache.hadoop.hbase.HColumnDescriptor;
9import org.apache.hadoop.hbase.HTableDescriptor;
10import org.apache.hadoop.hbase.client.HBaseAdmin;
11import org.apache.hadoop.hbase.client.HTable;
12import org.apache.hadoop.hbase.client.Put;
13import org.apache.hadoop.hbase.client.Scan;
14import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
15import org.apache.hadoop.hbase.mapreduce.TableMapper;
16import org.apache.hadoop.hbase.mapreduce.TableReducer;
17import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
18import org.apache.hadoop.hbase.client.Result;
19// Hadoop classes
20import org.apache.hadoop.util.ToolRunner;
21import org.apache.hadoop.conf.Configured;
22import org.apache.hadoop.conf.Configuration;
23import org.apache.hadoop.util.Tool;
24import org.apache.hadoop.io.IntWritable;
25import org.apache.hadoop.io.Text;
26import org.apache.hadoop.mapreduce.Job;
27import org.apache.hadoop.mapreduce.Reducer.Context;
28
29public class Selection extends Configured implements Tool {
30 private static String inputTable;
31 private static String outputTable;
32
33 // =================================================================== Main
34 /*
35 * Explanation: This MapReduce job either requires at input both family and
36 * column name defined (family:column), or if only the column name is
37 * provided, it assumes that both family and column name are the same. For
38 * example, for a table with columns a, b and c, it would assume three
39 * families (a, b and c).
40 *
41 * This MapReduce takes three parameters: - Input HBase table from where to
42 * read data. - Output HBase table where to store data. - A list of
43 * [family:]columns to project.
44 *
45 * We distinguish two following cases: 1) For example, assume the following
46 * HBase table UsernameInput (the corresponding shell create statement
47 * follows): create 'UsernameInput', 'a', 'b' -- It contains two families: a
48 * and b put 'UsernameInput', 'key1', 'a:a', '1' -- It creates an attribute
49 * a under the a family with value 1 put 'UsernameInput', 'key1', 'b:b', '2'
50 * -- It creates an attribute b under the b family with value 2
51 *
52 * A correct call would be this: yarn jar myJarFile.jar Projection
53 * UsernameInput out [a:]a -- It projects a The result (stored in
54 * UsernameOutput) would be: 'key1', 'a:a', '1' -- b is not there Notice
55 * that in this case providing family name is optional.
56 *
57 * 2) However, assume the following case where HBase table is created as
58 * follows: create 'UsernameInputF', 'cf1', 'cf2' -- It contains two
59 * families: cf1 and cf2 put 'UsernameInputF', 'key1', 'cf1:a', '1' -- It
60 * creates an attribute a under the cf1 family with value 1 put
61 * 'UsernameInputF', 'key1', 'cf2:b', '2' -- It creates an attribute b under
62 * the cf2 family with value 2
63 *
64 * In this case, a correct call would require both family and column
65 * defined, as follows: yarn jar myJarFile.jar Projection UsernameInputF
66 * UsernameOutputF cf1:a -- It projects cf1:a The result (stored in
67 * UsernameOutputF) would be: 'key1', 'cf1:a', '1' -- cf2:b is not there
68 * Notice that in this case providing family name is mandatory.
69 *
70 */
71
72 public static void main(String[] args) throws Exception {
73 if (args.length < 4) {
74 System.err.println("Parameters missing: 'inputTable outputTable [family:]attribute value*'");
75 System.exit(1);
76 }
77 inputTable = args[0];
78 outputTable = args[1];
79
80 int tablesRight = checkIOTables(args);
81 if (tablesRight == 0) {
82 int ret = ToolRunner.run(new Selection(), args);
83 System.exit(ret);
84 } else {
85 System.exit(tablesRight);
86 }
87 }
88
89 // =============================================================
90 // checkTables
91 private static int checkIOTables(String[] args) throws Exception {
92 // Obtain HBase's configuration
93 Configuration config = HBaseConfiguration.create();
94 // Create an HBase administrator
95 HBaseAdmin hba = new HBaseAdmin(config);
96
97 // With an HBase administrator we check if the input table exists
98 if (!hba.tableExists(inputTable)) {
99 System.err.println("Input table does not exist");
100 return 2;
101 }
102 // Check if the output table exists
103 if (hba.tableExists(outputTable)) {
104 System.err.println("Output table already exists");
105 return 3;
106 }
107 HTableDescriptor htdInput = hba.getTableDescriptor(inputTable.getBytes());
108
109 // Create the columns of the output table
110 HTableDescriptor htdOutput = new HTableDescriptor(outputTable.getBytes());
111 // Add columns to the new table
112 System.out.println("Comencem");
113 for (byte[] family : htdInput.getFamiliesKeys()) {
114 System.out.println(family);
115 htdOutput.addFamily(new HColumnDescriptor(family));
116 }
117
118 // If you want to insert data do it here
119 // -- Inserts
120 // -- Inserts
121 // Create the new output table
122 hba.createTable(htdOutput);
123 return 0;
124 }
125
126 // ============================================================== Job config
127 public int run(String [] args) throws Exception {
128 //Create a new job to execute
129
130 //Retrive the configuration
131 Job job = new Job(HBaseConfiguration.create());
132 //Set the MapReduce class
133 job.setJarByClass(Selection.class);
134 //Set the job name
135 job.setJobName("Selection");
136 //Create an scan object
137 Scan scan = new Scan();
138 //Set the columns to scan and keep header to project
139 String[] familyColumn = new String[2];
140 if (args[2].contains(":")) {
141 familyColumn = args[2].split(":");
142 } else {
143 familyColumn[0] = args[2];
144 familyColumn[1] = args[2];
145 }
146 String value = args[3];
147
148 scan.addColumn(familyColumn[0].getBytes(), familyColumn[1].getBytes());
149 String header = familyColumn + "," + value;
150
151 job.getConfiguration().setStrings("attributes",header);
152 //Set the Map and Reduce function
153 TableMapReduceUtil.initTableMapperJob(inputTable, scan, Mapper.class, Text.class, Text.class, job);
154 TableMapReduceUtil.initTableReducerJob(outputTable, Reducer.class, job);
155
156 boolean success = job.waitForCompletion(true);
157 return success ? 0 : 4;
158 }
159
160 // ===================================================================
161 // Mapper
162 public static class Mapper extends TableMapper<Text, Text> {
163
164 public void map(ImmutableBytesWritable rowMetadata, Result values, Context context)
165 throws IOException, InterruptedException {
166 String tuple = "";
167 String val = "";
168 String rowId = new String(rowMetadata.get(), "US-ASCII");
169 String[] attributes = context.getConfiguration().getStrings("attributes", "empty");
170
171 // familyColumn contains [family:]column we are looking for
172 String familyColumn = attributes[0].split(",")[0];
173 // value contains the value we are looking for
174 String value = attributes[0].split(",")[1];
175
176 String[] firstFamilyColumn = new String[2];
177 if (!familyColumn.contains(":")) {
178 // If only the column name is provided, it is assumed that both
179 // family and column names are the same
180 firstFamilyColumn[0] = familyColumn;
181 firstFamilyColumn[1] = familyColumn;
182 } else {
183 firstFamilyColumn = familyColumn.split(":");
184 }
185
186 if (values.containsColumn(firstFamilyColumn[0].getBytes(), firstFamilyColumn[1].getBytes())) {
187 val = new String(values.getValue(firstFamilyColumn[0].getBytes(), firstFamilyColumn[1].getBytes()));
188 if (val.equals(value))
189 tuple = values.toString();
190 context.write(new Text(tuple), new Text(rowId));
191 }
192 }
193 }
194
195 // ==================================================================
196 // Reducer
197 public static class Reducer extends TableReducer<Text, Text, Text> {
198
199 public void reduce(Text key, Iterable<Text> inputList, Context context)
200 throws IOException, InterruptedException {
201 Iterator<Text> iterator = inputList.iterator();
202 while (iterator.hasNext()) {
203 Text outputKey = inputList.iterator().next();
204 //
205 //String row_id = outputKey.toString().split(",")[0];
206 //String tupla = outputKey.toString().split(",")[1];
207 // Create a tuple for the output table
208 Put put = new Put(outputKey.getBytes());
209 // Set the values for the columns
210 String[] attributes = context.getConfiguration().getStrings("attributes", "empty");
211 String[] familyColumn = attributes[0].split(",");
212
213 put.add(familyColumn[0].getBytes(), familyColumn[1].getBytes(), key.getBytes());
214 // Put the tuple in the output table
215 context.write(outputKey, put);
216
217 }
218 }
219 }
220}