· 9 years ago · Oct 31, 2016, 05:48 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;
19import org.apache.hadoop.hbase.KeyValue;
20
21// Hadoop classes
22import org.apache.hadoop.util.ToolRunner;
23import org.apache.hadoop.conf.Configured;
24import org.apache.hadoop.conf.Configuration;
25import org.apache.hadoop.util.Tool;
26import org.apache.hadoop.io.IntWritable;
27import org.apache.hadoop.io.Text;
28import org.apache.hadoop.mapreduce.Job;
29import org.apache.hadoop.mapreduce.Reducer.Context;
30
31public class Selection extends Configured implements Tool {
32 private static String inputTable;
33 private static String outputTable;
34
35 public static void main(String[] args) throws Exception {
36 if (args.length < 4) {
37 System.err.println("Parameters missing: 'inputTable outputTable [family:]attribute value*'");
38 System.exit(1);
39 }
40 inputTable = args[0];
41 outputTable = args[1];
42
43 int tablesRight = checkIOTables(args);
44 if (tablesRight == 0) {
45 int ret = ToolRunner.run(new Selection(), args);
46 System.exit(ret);
47 } else {
48 System.exit(tablesRight);
49 }
50 }
51
52 // =============================================================
53 // checkTables
54 private static int checkIOTables(String[] args) throws Exception {
55 // Obtain HBase's configuration
56 Configuration config = HBaseConfiguration.create();
57 // Create an HBase administrator
58 HBaseAdmin hba = new HBaseAdmin(config);
59
60 // With an HBase administrator we check if the input table exists
61 if (!hba.tableExists(inputTable)) {
62 System.err.println("Input table does not exist");
63 return 2;
64 }
65 // Check if the output table exists
66 if (hba.tableExists(outputTable)) {
67 System.err.println("Output table already exists");
68 return 3;
69 }
70 // Get table from inputTable
71 HTableDescriptor htdInput = hba.getTableDescriptor(inputTable.getBytes());
72
73 // Create the columns of the output table
74 HTableDescriptor htdOutput = new HTableDescriptor(outputTable.getBytes());
75 // Add columns to the new table
76 for (byte[] family : htdInput.getFamiliesKeys()) {
77 htdOutput.addFamily(new HColumnDescriptor(family));
78 }
79
80 // If you want to insert data do it here
81 // -- Inserts
82 // -- Inserts
83 // Create the new output table
84 hba.createTable(htdOutput);
85 return 0;
86 }
87
88 // ============================================================== Job config
89 public int run(String[] args) throws Exception {
90 //Create a new job to execute
91
92 //Retrive the configuration
93 Job job = new Job(HBaseConfiguration.create());
94 //Set the MapReduce class
95 job.setJarByClass(Selection.class);
96 //Set the job name
97 job.setJobName("Selection");
98 //Create an scan object
99 Scan scan = new Scan();
100
101 String familyColumn = args[2];
102 String value = args[3];
103
104 String header = familyColumn + "," + args[3];
105 job.getConfiguration().setStrings("attributes", header);
106 //Set the Map and Reduce function
107 TableMapReduceUtil.initTableMapperJob(inputTable, scan, Mapper.class, Text.class, Text.class, job);
108 TableMapReduceUtil.initTableReducerJob(outputTable, Reducer.class, job);
109
110 boolean success = job.waitForCompletion(true);
111 return success ? 0 : 4;
112 }
113
114 // ===================================================================
115 // Mapper
116 public static class Mapper extends TableMapper<Text, Text> {
117
118 public void map(ImmutableBytesWritable rowMetadata, Result values, Context context)
119 throws IOException, InterruptedException {
120 String rowId = new String(rowMetadata.get(), "US-ASCII");
121 String[] attributes = context.getConfiguration().getStrings("attributes", "empty");
122
123 // attribute is the column we are looking for
124 String familyColumn = attributes[0];
125 // value is the value of the attribute we are looking for
126 String value = attributes[1];
127
128 // columnValue contains the value of the column we might be looking for
129 String columnValue;
130 if (familyColumn.contains(":")) {
131 // The family has been defined as family:attribute
132 String[] familyColumnArray = familyColumn.split(":");
133 columnValue = new String(values.getValue(familyColumnArray[0].getBytes(), familyColumnArray[1].getBytes()));
134 } else {
135 // The family is the same as the attribute
136 columnValue = new String(values.getValue(familyColumn.getBytes(), familyColumn.getBytes()));
137 }
138
139 // If the value of the column is equals to the value we are looking for, write the content of the key to context
140 if (columnValue.equals(value)) {
141 KeyValue[] raw = values.raw();
142 String tuple = new String(raw[0].getFamily()) + ":" + new String(raw[0].getValue());
143 for (int i = 1; i < raw.length; i++)
144 tuple += "," + new String(raw[i].getFamily()) + ":" + new String(raw[i].getValue());
145 // Send to reducer tuple identified by the 'rowId'
146 context.write(new Text(rowId), new Text(tuple));
147 }
148
149 // In conclusion, we send to reducer the rows that attribute value is equals
150 // with value received as a parameter
151 }
152 }
153
154 // ==================================================================
155 // Reducer
156 public static class Reducer extends TableReducer<Text, Text, Text> {
157
158 public void reduce(Text key, Iterable<Text> inputList, Context context)
159 throws IOException, InterruptedException {
160 Iterator<Text> iterator = inputList.iterator();
161 while (inputList.iterator().hasNext()) {
162 // 'inputList' contains de rows that achieve the condition.
163 // Iterate for all of them adding in the output table.
164 Text outputKey = inputList.iterator().next();
165 Put put = new Put(key.getBytes());
166 for (String row : outputKey.toString().split(",")) {
167 String[] values = row.split(":");
168 // Adding the family, qualifier and the value respectively.
169 put.add(values[0].getBytes(), values[0].getBytes(), values[1].getBytes());
170 }
171 // Write to output table.
172 context.write(outputKey, put);
173 }
174 }
175 }
176}