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) {

   }

}

a1.sources = r1

a1.channels = c1 c2

a1.sinks = k1 sink2

a1.source.s1.selector.type = replicating

#NetCat TCP source

a1.sources.r1.type = netcat

a1.sources.r1.bind = 0.0.0.0

a1.sources.r1.port = 6666

a1.sources.r1.channels = c1 c2

#channel

a1.channels.c2.type = memory

a1.channels.c2.capacity = 10000

a1.channels.c2.transactionCapacity = 1000

#HBase sink

a1.sinks.sink2.type = org.apache.flume.sink.hbase.AsyncHBaseSink

a1.sinks.sink2.channel = c2

a1.sinks.sink2.table = law

a1.sinks.sink2.columnFamily = lawfile

a1.sinks.sink2.batchSize = 5000

#The serializer to use

a1.sinks.sink2.serializer = ifre.flume.hbase.SplittingSerializer

#List of columns each event writes to.

a1.sinks.sink2.serializer.columns = name,id,phone

# hbase shell

create "law" "lawfile"

[root@ifrebigsearch2 apache-flume-1.6.0-bin]# bin/flume-ng agent --conf conf --conf-file conf/crawler-hdfs-conf.properties --name a1 -Dflume.root.logger=INFO,console

[root@ifrebigsearch0 dkh]# nc ifrebigsearch2 6666

zhangsan,10110198806054561,13812345678

OK

hbase(main):002:0> scan 'law'

ROW                   COLUMN+CELL                                              

9223370524386395508z column=lawfile:id, timestamp=1512468380362, value=10110198

hangsan              806054561                                                

9223370524386395508z column=lawfile:name, timestamp=1512468380361, value=zhangs

hangsan              an                                                        

9223370524386395508z column=lawfile:phone, timestamp=1512468380363, value=13812

hangsan              345678