· 9 years ago · Oct 31, 2016, 02:18 PM
1package cbde.lab4;
2
3// Standard classes
4import java.io.IOException;
5import java.util.Vector;
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;
19import org.apache.hadoop.hbase.util.Bytes;
20
21
22// Hadoop classes
23import org.apache.hadoop.util.ToolRunner;
24import org.apache.hadoop.conf.Configured;
25import org.apache.hadoop.conf.Configuration;
26import org.apache.hadoop.util.Tool;
27import org.apache.hadoop.io.IntWritable;
28import org.apache.hadoop.io.Text;
29import org.apache.hadoop.mapreduce.Job;
30import org.apache.hadoop.mapreduce.Reducer.Context;
31
32
33public class GroupBy extends Configured implements Tool {
34 private static String inputTable;
35 private static String outputTable;
36
37//=================================================================== Main
38 /*
39
40 This MapReduce simulates the Group By operation in a given table for a single attribute considering by default the
41 SUM aggregation for a signle attribute
42
43 This MapReduce takes four parameters:
44 - Input HBase table from where to read data.
45 - Output HBase table where to store data.
46 - An aggregate attribute
47 - A group by attribute
48
49 We distinguish two following cases:
50 1) For example, assume the following HBase table UsernameInput (the corresponding shell create statement follows):
51 create 'UsernameInput', 'a', 'b' -- It contains two families: a and b
52 put 'UsernameInput', 'key1', 'a:a', '1' -- It creates an attribute a under the a family with value 1
53 put 'UsernameInput', 'key1', 'b:b', '2' -- It creates an attribute b under the b family with value 2
54 put 'UsernameInput', 'key2', 'a:a', '1' -- Adds a new value to the attribute a of family a with value 1
55 put 'UsernameInput', 'key2', 'b:b', '1' -- Adds a new value to the attribute b of family b with value 1
56
57 A correct call would be this: yarn jar myJarFile.jar GroupBy UsernameInput out [b:]b [a:]a -- It groups by a
58 and aggregates by b (SUM(b))
59 The result (stored in UsernameOutput) would be: 'a:a', '1'; 'b:b', '3'
60 Notice that in this case providing family name is optional.
61
62 2) However, assume the following case where HBase table is created as follows:
63 create 'UsernameInputF', 'cf1', 'cf2' -- It contains two families: cf1 and cf2
64 put 'UsernameInputF', 'key1', 'cf1:a', '1' -- It creates an attribute a under the cf1 family with value 1
65 put 'UsernameInputF', 'key1', 'cf2:b', '2' -- It creates an attribute b under the cf2 family with value 2
66 put 'UsernameInputF', 'key2', 'cf1:a', '1' -- Adds a new value to the attribute a of family cf1 with value 1
67 put 'UsernameInputF', 'key2', 'cf2:b', '1' -- Adds a new value to the attribute b of family cf2 with value 1
68
69 In this case, a correct call would require both family and column defined, as follows:
70 yarn jar myJarFile.jar GroupBy UsernameInputF UsernameOutputF cf1:b cf1:a -- It groups by a and aggregates
71 by b (SUM(b))
72 The result (stored in UsernameOutputF) would be: 'cf1:a', '1'; 'cf2:b', '3'
73 Notice that in this case providing family name is mandatory.
74
75 */
76
77 public static void main(String[] args) throws Exception {
78 if (args.length<4) {
79 System.err.println("Parameters missing: 'inputTable outputTable aggregateAttribute groupByAttribute'");
80 System.exit(1);
81 }
82 inputTable = args[0];
83 outputTable = args[1];
84
85 int tablesRight = checkIOTables(args);
86 if (tablesRight==0) {
87 int ret = ToolRunner.run(new GroupBy(), args);
88 System.exit(ret);
89 } else {//Add columns to the new table
90 System.exit(tablesRight);
91 }
92 }
93
94
95//============================================================== checkTables
96 private static int checkIOTables(String [] args) throws Exception {
97 // Obtain HBase's configuration
98 Configuration config = HBaseConfiguration.create();
99 // Create an HBase administrator
100 HBaseAdmin hba = new HBaseAdmin(config);
101
102 // With an HBase administrator we check if the input table exists
103 if (!hba.tableExists(inputTable)) {
104 System.err.println("Input table does not exist");
105 return 2;
106 }
107 // Check if the output table exists
108 if (hba.tableExists(outputTable)) {
109 System.err.println("Output table already exists");
110 return 3;
111 }
112 /*HTable table = new HTable(config, inputTable);
113 //You may not need this, but if batch loading, set the recovery manager to Not Force (setAutoFlush false) and deactivate the WAL
114 Put put = new Put(Bytes.toBytes("key1")); //creates a new row with key 'key1'
115 put.add(Bytes.toBytes("cf1"), Bytes.toBytes("a"), Bytes.toBytes(1)); //Add an attribute named Attribute that belongs to the family Family with value Value
116 put.add(Bytes.toBytes("cf2"), Bytes.toBytes("b"), Bytes.toBytes(2));
117 table.put(put); // Inserts data
118 Put put2 = new Put(Bytes.toBytes("key2")); //creates a new row with key 'key1'
119 put2.add(Bytes.toBytes("cf1"), Bytes.toBytes("a"), Bytes.toBytes(1)); //Add an attribute named Attribute that belongs to the family Family with value Value
120 put2.add(Bytes.toBytes("cf2"), Bytes.toBytes("b"), Bytes.toBytes(3));
121 table.put(put2); // Inserts data
122 */
123
124 //If you do it through the HBase Shell, this is equivalent to:
125 //put 'Username_InputTable1', 'key1', 'Family:Attribute', 'Value'
126 // Create the columns of the output table
127 HTableDescriptor htdOutput = new HTableDescriptor(outputTable.getBytes());
128 //Add columns to the new table
129 String[] aggregateFamilyColumn = new String[2];
130 if (!args[2].contains(":")){
131 //If only the column name is provided, it is assumed that both family and column names are the same
132 aggregateFamilyColumn[0] = args[2];
133 aggregateFamilyColumn[1] = args[2];
134 }
135 else {
136 //Otherwise, we extract family and column names from the provided argument "family:column"
137 aggregateFamilyColumn = args[2].split(":");
138 }
139 String[] groupByFamilyColumn = new String[2];
140 if (!args[3].contains(":")) {
141 //If only the column name is provided, it is assumed that both family and column names are the same
142 groupByFamilyColumn[0] = args[3];
143 groupByFamilyColumn[1] = args[3];
144 }
145 else {
146 //Otherwise, we extract family and column names from the provided argument "family:column"
147 groupByFamilyColumn = args[3].split(":");
148 }
149 htdOutput.addFamily(new HColumnDescriptor(groupByFamilyColumn[0]));
150 if (!groupByFamilyColumn[0].equals(aggregateFamilyColumn[0])) htdOutput.addFamily(new HColumnDescriptor(aggregateFamilyColumn[0]));
151
152 //Create the new output table
153 hba.createTable(htdOutput);
154 return 0;
155 }
156
157//============================================================== Job config
158 public int run(String [] args) throws Exception {
159 //Create a new job to execute
160
161 //Retrive the configuration
162 Job job = new Job(HBaseConfiguration.create());
163 //Set the MapReduce class
164 job.setJarByClass(GroupBy.class);
165 //Set the job name
166 job.setJobName("GroupBy");
167 //Create an scan object
168 Scan scan = new Scan();
169
170 String[] aggregateFamilyColumn = new String[2];
171 if (!args[2].contains(":")){
172 //If only the column name is provided, it is assumed that both family and column names are the same
173 aggregateFamilyColumn[0] = args[2];
174 aggregateFamilyColumn[1] = args[2];
175 }
176 else {
177 //Otherwise, we extract family and column names from the provided argument "family:column"
178 aggregateFamilyColumn = args[2].split(":");
179 }
180 String[] groupByFamilyColumn = new String[2];
181 if (!args[3].contains(":")) {
182 //If only the column name is provided, it is assumed that both family and column names are the same
183 groupByFamilyColumn[0] = args[3];
184 groupByFamilyColumn[1] = args[3];
185 }
186 else {
187 //Otherwise, we extract family and column names from the provided argument "family:column"
188 groupByFamilyColumn = args[3].split(":");
189 }
190
191
192 String header = aggregateFamilyColumn[0] + ":" + aggregateFamilyColumn[1];
193 header = header + "," + groupByFamilyColumn[0] + ":" + groupByFamilyColumn[1];
194 job.getConfiguration().setStrings("header", header);
195 //Set the Map and Reduce function
196 TableMapReduceUtil.initTableMapperJob(inputTable, scan, Mapper.class, Text.class, Text.class, job);
197 TableMapReduceUtil.initTableReducerJob(outputTable, Reducer.class, job);
198 boolean success = job.waitForCompletion(true);
199 return success ? 0 : 4;
200 }
201
202
203//=================================================================== Mapper
204 public static class Mapper extends TableMapper<Text, Text> {
205
206 public void map(ImmutableBytesWritable rowMetadata, Result values, Context context) throws IOException, InterruptedException {
207 String value = "";
208 String key = "";
209
210 String[] header = context.getConfiguration().getStrings("header");
211
212 String[] aggregate = header[0].split(":");
213 String[] groupBy = header[1].split(":");
214
215 key = new String(values.getValue(groupBy[0].getBytes(),groupBy[1].getBytes()));
216 value = new String(values.getValue(aggregate[0].getBytes(),aggregate[1].getBytes()));
217
218 context.write(new Text(key), new Text(value));
219 }
220 }
221
222//================================================================== Reducer
223 public static class Reducer extends TableReducer<Text, Text, Text> {
224
225 public void reduce(Text key, Iterable<Text> inputList, Context context) throws IOException, InterruptedException {
226
227 String[] header = context.getConfiguration().getStrings("header");
228 String[] aggregate = header[0].split(":");
229 String[] groupBy = header[1].split(":");
230 String rowKey = groupBy[0] + ":" + groupBy[1] + "-" + key.toString();
231 Put put = new Put(rowKey.getBytes());
232
233 int sum = 0;
234 while (inputList.iterator().hasNext()) {
235 Text val = inputList.iterator().next();
236 sum = sum + Integer.parseInt(val.toString());
237 }
238 put.add(aggregate[0].getBytes(), aggregate[1].getBytes(), Integer.toString(sum).getBytes());
239 // Put the tuple in the output table
240 context.write(new Text(rowKey), put);
241 }
242
243 }
244 }