· 9 years ago · Oct 31, 2016, 06:42 PM
1import java.io.IOException;
2import java.util.ArrayList;
3import org.apache.hadoop.conf.Configuration;
4import org.apache.hadoop.conf.Configured;
5import org.apache.hadoop.hbase.HBaseConfiguration;
6import org.apache.hadoop.hbase.HColumnDescriptor;
7import org.apache.hadoop.hbase.HTableDescriptor;
8import org.apache.hadoop.hbase.client.HBaseAdmin;
9import org.apache.hadoop.hbase.client.Put;
10import org.apache.hadoop.hbase.client.Result;
11import org.apache.hadoop.hbase.client.Scan;
12import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
13import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
14import org.apache.hadoop.hbase.mapreduce.TableMapper;
15import org.apache.hadoop.hbase.mapreduce.TableReducer;
16import org.apache.hadoop.hbase.util.Bytes;
17import org.apache.hadoop.io.Text;
18import org.apache.hadoop.mapreduce.Job;
19import org.apache.hadoop.util.Tool;
20import org.apache.hadoop.util.ToolRunner;
21
22// EXAMPLE OF JOIN
23
24// INPUT
25
26// QUERY
27
28// OUTPUT
29
30public class Join extends Configured implements Tool {
31 public static final String PARAMETERS = "'leftInputTable rightInputTable outputTable leftAttribute rightAttribute'";
32 public static final String JOB_NAME = "Join";
33 public static final String ATTRIBUTES = "attributes";
34
35 private static String inputTable1;
36 private static String inputTable2;
37 private static String outputTable;
38
39 public static void main(String[] args) throws Exception {
40 // Check the quantity of params received.
41 if (args.length != 5) {
42 System.err.println("Parameters missing: " + PARAMETERS);
43 System.exit(1);
44 }
45 // Assign the input and output tables to global variables.
46 inputTable1 = args[0];
47 inputTable2 = args[1];
48 outputTable = args[2];
49
50 // Check the validity tables.
51 int tablesRight = checkIOTables(args);
52 if (tablesRight == 0) {
53 // Execute the algorithm.
54 int ret = ToolRunner.run(new Join(), args);
55 System.exit(ret);
56 } else {
57 System.exit(tablesRight);
58 }
59 }
60
61 private static int checkIOTables(String [] args) throws Exception {
62 Configuration config = HBaseConfiguration.create();
63 HBaseAdmin hba = new HBaseAdmin(config);
64
65 // Check the existence of the input table.
66 if (!hba.tableExists(inputTable1)) {
67 System.err.println("Input table 1 does not exist");
68 return 2;
69 }
70 // Check the existence of the input table.
71 if (!hba.tableExists(inputTable2)) {
72 System.err.println("Input table 2 does not exist");
73 return 2;
74 }
75 // Check the nonexistence of the output table.
76 if (hba.tableExists(outputTable)) {
77 System.err.println("Output table already exists");
78 return 3;
79 }
80
81 // Initialize the outputTable
82 HTableDescriptor htdOutput = new HTableDescriptor(outputTable.getBytes());
83
84 // Assign the same families that left input table.
85 HTableDescriptor htdInput1 = hba.getTableDescriptor(inputTable1.getBytes());
86 for (byte[] familyKey : htdInput1.getFamiliesKeys())
87 htdOutput.addFamily(new HColumnDescriptor(familyKey));
88
89 // Assign the same families that right input table.
90 HTableDescriptor htdInput2 = hba.getTableDescriptor(inputTable2.getBytes());
91 for (byte[] familyKey : htdInput2.getFamiliesKeys())
92 htdOutput.addFamily(new HColumnDescriptor(familyKey));
93
94 hba.createTable(htdOutput);
95 return 0;
96 }
97
98 public int run(String [] args) throws Exception {
99 // Create Configuration.
100 Job job = new Job(HBaseConfiguration.create());
101 job.setJarByClass(Join.class);
102 job.setJobName(JOB_NAME);
103
104 // Create header and assign to configuration.
105 String header = args[3] + "," + args[4];
106 job.getConfiguration().setStrings(ATTRIBUTES, header);
107
108 ArrayList<Scan> scans = new ArrayList<Scan>();
109 Scan scan1 = new Scan();
110 scan1.setAttribute("scan.attributes.table.name", Bytes.toBytes(inputTable1));
111 scans.add(scan1);
112 Scan scan2 = new Scan();
113 scan2.setAttribute("scan.attributes.table.name", Bytes.toBytes(inputTable2));
114 scans.add(scan2);
115
116 // Init map and reduce functions
117 TableMapReduceUtil.initTableMapperJob(scans, Mapper.class, Text.class, Text.class, job);
118 TableMapReduceUtil.initTableReducerJob(outputTable, Reducer.class, job);
119
120 boolean success = job.waitForCompletion(true);
121 return success ? 0 : 4;
122 }
123
124 public static class Mapper extends TableMapper<Text, Text> {
125
126 public void map(ImmutableBytesWritable rowMetadata, Result values, Context context) throws IOException, InterruptedException {
127
128 String[] attributes = context.getConfiguration().getStrings(ATTRIBUTES, "empty");
129
130 }
131 }
132
133 public static class Reducer extends TableReducer<Text, Text, Text> {
134
135 public void reduce(Text key, Iterable<Text> inputList, Context context) throws IOException, InterruptedException {
136
137 String[] attributes = context.getConfiguration().getStrings(ATTRIBUTES, "empty");
138
139 }
140 }
141}