· 9 years ago · Oct 31, 2016, 04:42 PM
1import org.apache.hadoop.conf.Configuration;
2import org.apache.hadoop.conf.Configured;
3import org.apache.hadoop.hbase.HBaseConfiguration;
4import org.apache.hadoop.hbase.HColumnDescriptor;
5import org.apache.hadoop.hbase.HTableDescriptor;
6import org.apache.hadoop.hbase.KeyValue;
7import org.apache.hadoop.hbase.client.HBaseAdmin;
8import org.apache.hadoop.hbase.client.Put;
9import org.apache.hadoop.hbase.client.Result;
10import org.apache.hadoop.hbase.client.Scan;
11import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
12import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
13import org.apache.hadoop.hbase.mapreduce.TableMapper;
14import org.apache.hadoop.hbase.mapreduce.TableReducer;
15import org.apache.hadoop.io.Text;
16import org.apache.hadoop.mapreduce.Job;
17import org.apache.hadoop.util.Tool;
18import org.apache.hadoop.util.ToolRunner;
19import java.io.IOException;
20
21// EXAMPLE OF SELECTION
22
23// INPUT
24// ROW COLUMN+CELL
25// k1 column=a:a, timestamp=1477912921357, value=25
26// k1 column=b:b, timestamp=1477912921413, value=10
27// k1 column=c:c, timestamp=1477912921475, value=300
28// k2 column=a:a, timestamp=1477912921508, value=50
29// k2 column=b:b, timestamp=1477912921543, value=10
30// k3 column=a:a, timestamp=1477912921576, value=10
31// k3 column=b:b, timestamp=1477912923854, value=20
32
33// QUERY
34// yarn jar RA.jar Selection equipF_selectionIN equipF_selectionOUT b 10
35
36// OUTPUT
37// ROW COLUMN+CELL
38// k1 column=a:a, timestamp=1477913728071, value=25
39// k1 column=b:b, timestamp=1477913728071, value=10
40// k1 column=c:c, timestamp=1477913728071, value=300
41// k2 column=a:a, timestamp=1477913728071, value=50
42// k2 column=b:b, timestamp=1477913728071, value=10
43
44public class Selection extends Configured implements Tool {
45 public static final String PARAMETERS = "'inputTable outputTable [family:]attribute value'";
46 public static final String JOB_NAME = "Selection";
47 public static final String ATTRIBUTES = "attributes";
48 public static final String COLON = ":";
49 public static final String SEMI_COLON = ";";
50
51 private static String inputTable;
52 private static String outputTable;
53 private static String attribute;
54 private static String value;
55
56 public static void main(String[] args) throws Exception {
57 // Check the quantity of params received.
58 if (args.length != 4) {
59 System.err.println("Parameters missing: " + PARAMETERS);
60 System.exit(1);
61 }
62 // Assign the input, output table, attribute and value to global variables.
63 inputTable = args[0];
64 outputTable = args[1];
65 attribute = args[2];
66 value = args[3];
67
68 // Check the validity tables.
69 int tablesRight = checkIOTables(args);
70 if (tablesRight == 0) {
71 // Execute the algorithm.
72 int ret = ToolRunner.run(new Selection(), args);
73 System.exit(ret);
74 } else {
75 System.exit(tablesRight);
76 }
77 }
78
79 private static int checkIOTables(String [] args) throws Exception {
80 Configuration config = HBaseConfiguration.create();
81 HBaseAdmin hba = new HBaseAdmin(config);
82
83 // Check the existence of the input table.
84 if (!hba.tableExists(inputTable)) {
85 System.err.println("Input table does not exist");
86 return 2;
87 }
88 // Check the nonexistence of the output table.
89 if (hba.tableExists(outputTable)) {
90 System.err.println("Output table already exists");
91 return 3;
92 }
93
94 // Create the output table and assign the same families as input table.
95 HTableDescriptor htdInput = hba.getTableDescriptor(inputTable.getBytes());
96 HTableDescriptor htdOutput = new HTableDescriptor(outputTable.getBytes());
97 for (byte[] familyKey : htdInput.getFamiliesKeys())
98 htdOutput.addFamily(new HColumnDescriptor(familyKey));
99 hba.createTable(htdOutput);
100
101 return 0;
102 }
103
104 public int run(String [] args) throws Exception {
105 // Create Configuration.
106 Job job = new Job(HBaseConfiguration.create());
107 job.setJarByClass(Selection.class);
108 job.setJobName(JOB_NAME);
109
110 // Create header and assign to configuration.
111 Scan scan = new Scan();
112 String header = attribute + "," + value;
113 job.getConfiguration().setStrings(ATTRIBUTES, header);
114
115 // Init map and reduce functions
116 TableMapReduceUtil.initTableMapperJob(inputTable, scan, Mapper.class, Text.class, Text.class, job);
117 TableMapReduceUtil.initTableReducerJob(outputTable, Reducer.class, job);
118
119 boolean success = job.waitForCompletion(true);
120 return success ? 0 : 4;
121 }
122
123 public static class Mapper extends TableMapper<Text, Text> {
124
125 public void map(ImmutableBytesWritable rowMetadata, Result values, Context context) throws IOException, InterruptedException {
126
127 String rowId = new String(rowMetadata.get(), "US-ASCII");
128 String[] attributes = context.getConfiguration().getStrings(ATTRIBUTES, "empty");
129
130 // Get the attribute and the value to select.
131 String attribute = attributes[0];
132 String value = attributes[1];
133
134 // Assign to 'attributeValue' the value of the attribute to compare with the selection value
135 String attributeValue;
136 if (attribute.contains(COLON)) {
137 String[] split = attribute.split(COLON);
138 attributeValue = new String(values.getValue(split[0].getBytes(), split[1].getBytes()));
139 }
140 else
141 attributeValue = new String(values.getValue(attribute.getBytes(), attribute.getBytes()));
142
143 // If the value is equals, write the content of the key to context
144 if (attributeValue.equals(value)) {
145 KeyValue[] raw = values.raw();
146 String tuple = new String(raw[0].getFamily()) + COLON + new String(raw[0].getValue());
147 for (int i = 1; i < raw.length; i++)
148 tuple += SEMI_COLON + new String(raw[i].getFamily()) + COLON + new String(raw[i].getValue());
149 // Send to reducer tuple identified by the 'rowId'
150 context.write(new Text(rowId), new Text(tuple));
151 }
152
153 // In conclusion, we send to reducer the rows that attribute value is equals
154 // with value received as a parameter
155 }
156 }
157
158 public static class Reducer extends TableReducer<Text, Text, Text> {
159
160 public void reduce(Text key, Iterable<Text> inputList, Context context) throws IOException, InterruptedException {
161
162 while (inputList.iterator().hasNext()) {
163 // 'inputList' contains de rows that achieve the condition.
164 // Iterate for all of them adding in the output table.
165 Text outputKey = inputList.iterator().next();
166 Put put = new Put(key.getBytes());
167 for (String row : outputKey.toString().split(SEMI_COLON)) {
168 String[] values = row.split(COLON);
169 // Adding the family, qualifier and the value respectively.
170 put.add(values[0].getBytes(), values[0].getBytes(), values[1].getBytes());
171 }
172 // Write to output table.
173 context.write(outputKey, put);
174 }
175 }
176 }
177}