1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.client.example;
20
21 import org.apache.commons.logging.Log;
22 import org.apache.commons.logging.LogFactory;
23 import org.apache.hadoop.conf.Configured;
24 import org.apache.hadoop.hbase.TableName;
25 import org.apache.hadoop.hbase.client.BufferedMutator;
26 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
27 import org.apache.hadoop.hbase.client.Connection;
28 import org.apache.hadoop.hbase.client.ConnectionFactory;
29 import org.apache.hadoop.hbase.client.Put;
30 import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
31 import org.apache.hadoop.hbase.util.Bytes;
32 import org.apache.hadoop.util.Tool;
33 import org.apache.hadoop.util.ToolRunner;
34
35 import java.io.IOException;
36 import java.util.ArrayList;
37 import java.util.List;
38 import java.util.concurrent.Callable;
39 import java.util.concurrent.ExecutionException;
40 import java.util.concurrent.ExecutorService;
41 import java.util.concurrent.Executors;
42 import java.util.concurrent.Future;
43 import java.util.concurrent.TimeUnit;
44 import java.util.concurrent.TimeoutException;
45
46
47
48
49 public class BufferedMutatorExample extends Configured implements Tool {
50
51 private static final Log LOG = LogFactory.getLog(BufferedMutatorExample.class);
52
53 private static final int POOL_SIZE = 10;
54 private static final int TASK_COUNT = 100;
55 private static final TableName TABLE = TableName.valueOf("foo");
56 private static final byte[] FAMILY = Bytes.toBytes("f");
57
58 @Override
59 public int run(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
60
61
62 final BufferedMutator.ExceptionListener listener = new BufferedMutator.ExceptionListener() {
63 @Override
64 public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator mutator) {
65 for (int i = 0; i < e.getNumExceptions(); i++) {
66 LOG.info("Failed to sent put " + e.getRow(i) + ".");
67 }
68 }
69 };
70 BufferedMutatorParams params = new BufferedMutatorParams(TABLE)
71 .listener(listener);
72
73
74
75
76 try (final Connection conn = ConnectionFactory.createConnection(getConf());
77 final BufferedMutator mutator = conn.getBufferedMutator(params)) {
78
79
80 final ExecutorService workerPool = Executors.newFixedThreadPool(POOL_SIZE);
81 List<Future<Void>> futures = new ArrayList<>(TASK_COUNT);
82
83 for (int i = 0; i < TASK_COUNT; i++) {
84 futures.add(workerPool.submit(new Callable<Void>() {
85 @Override
86 public Void call() throws Exception {
87
88
89
90
91 Put p = new Put(Bytes.toBytes("someRow"));
92 p.addColumn(FAMILY, Bytes.toBytes("someQualifier"), Bytes.toBytes("some value"));
93 mutator.mutate(p);
94
95
96 return null;
97 }
98 }));
99 }
100
101
102
103
104 for (Future<Void> f : futures) {
105 f.get(5, TimeUnit.MINUTES);
106 }
107 workerPool.shutdown();
108 } catch (IOException e) {
109
110 LOG.info("exception while creating/destroying Connection or BufferedMutator", e);
111 }
112
113 return 0;
114 }
115
116 public static void main(String[] args) throws Exception {
117 ToolRunner.run(new BufferedMutatorExample(), args);
118 }
119 }