importorg.apache.flume.Context; importorg.apache.flume.Event; importorg.apache.flume.conf.ComponentConfiguration; importorg.apache.flume.sink.hbase.AsyncHbaseEventSerializer; importorg.hbase.async.AtomicIncrementRequest; importorg.hbase.async.PutRequest; importjava.util.ArrayList; importjava.util.List; /** * Created by root on 12/5/17. */ public classSplittingSerializerimplementsAsyncHbaseEventSerializer { private byte[]table; private byte[]colFam; privateEventcurrentEvent; private byte[][]rentRowKey; private final byte[]eventCountCol="eventCount".getBytes(); columnNames; private finalListputs=newArrayList(); private finalListincs=newArrayList(); private byte[] cur public voidinitialize(byte[] table, byte[] cf) { this.table= table; this.colFam= cf; //Can not get the columns from context in configure method. Had to hard coded here. columnNames =new byte[3][]; columnNames[0] ="name".getBytes(); columnNames[1] ="id".getBytes(); columnNames[2] ="phone".getBytes(); } public voidsetEvent(Event event) { // Set the event and verify that the rowKey is not present this.currentEvent= event; /* //Don't know how to set the key of event header. String rowKeyStr = currentEvent.getHeaders().get("rowKey"); if (rowKeyStr == null) { throw new FlumeException("No row key found in headers!"); } currentRowKey = rowKeyStr.getBytes();*/ } publicListgetActions() { // Split the event body and get the values for the columns String eventStr =newString(currentEvent.getBody()); String[] cols = eventStr.split(","); Long currTime = System.currentTimeMillis(); longrevTs = Long.MAX_VALUE- currTime; currentRowKey = (Long.toString(revTs) + cols[0]).getBytes(); puts.clear(); for(inti =0;i < cols.length;i++) { //Generate a PutRequest for each column. PutRequest req =newPutRequest(table,currentRowKey,colFam, columnNames[i],cols[i].getBytes()); puts.add(req); } returnputs; } publicListgetIncrements() { incs.clear(); //Increment the number of events received incs.add(newAtomicIncrementRequest(table,"totalEvents".getBytes(),colFam,eventCountCol)); returnincs; } public voidcleanUp() { table=null; colFam=null; currentEvent=null; columnNames =null; currentRowKey =null; } public voidconfigure(Context context) { //Get the column names from the configuration //Did not work. Don't know how to use it. String cols =newString(context.getString("columns")); String[] names = cols.split(","); byte[][] columnNames =new byte[names.length][]; inti =0; System.out.println("getting columnNames"); for(String name : names) { columnNames[i++] = name.getBytes(); } } public voidconfigure(ComponentConfiguration componentConfiguration) { } } | 
