Copy Record From Current Date To New Date

Copy Record From Current Date To New Date



package copyrecord;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
import java.util.StringTokenizer;

/**
 * Created with IntelliJ IDEA.
 * User: Hadoop Share
 * Date: 2/12/13
 * Time: 11:58 AM
 * To change this template use File | Settings | File Templates.
 */
public class CopyRecordFromCurrentDateToNewDate {
    private static Configuration configuration = HBaseConfiguration.create();
    static{
        configuration.set("hbase.zookeeper.quorum","localhost");
        configuration.set("hbase.zookeeper.property.port","2181");
    }

    private void copyRecords(String tableName, String fromDate, String toDate) throws IOException {
        System.out.println("Begin Copy");
        HTableInterface table = new HTable(configuration,tableName);
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(fromDate));
        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner resultScanner = table.getScanner(scan);
        List<Put> putList = new ArrayList<Put>();
        for(Result result:resultScanner){
            System.out.println("  Row Key To Copy: " + Bytes.toString(result.getRow()));
            StringTokenizer tokenizer = new StringTokenizer(Bytes.toString(result.getRow()),":");
            tokenizer.nextToken();//skip date
            String newKey = toDate + tokenizer.nextToken();
            if(tokenizer.hasMoreTokens()){
                newKey = newKey  + ":" + tokenizer.nextToken();
            }
            Put put = new Put(newKey.getBytes());
            System.out.println("Row Key to Insert: " + Bytes.toString(put.getRow()));
            NavigableMap<byte[], NavigableMap<byte[], byte[]>> familyCellMap = result.getNoVersionMap();
            for(byte[] family:familyCellMap.keySet()){
                NavigableMap<byte[], byte[]> cell = familyCellMap.get(family);
                for (byte[] qualifier: cell.keySet()){
                   // System.out.println("Family: " + Bytes.toString(family) + " Qualifier: " + Bytes.toString(qualifier) + " Value: " + Bytes.toString(cell.get(qualifier)));
                    put.add(family,qualifier,cell.get(qualifier));
                }
            }
            putList.add(put);
        }

       // table.put(putList);
        System.out.println("End of Copy");

        //scan 'TRUNK_DETAIL', {FILTER => org.apache.hadoop.hbase.filter.RowFilter.new(CompareFilter::CompareOp.valueOf('EQUAL'),SubstringComparator.new("20131101:"))}
    }

    public static void main(String[] args) throws IOException {
        CopyRecordFromCurrentDateToNewDate copyRecordFromDateToDate = new CopyRecordFromCurrentDateToNewDate();
        copyRecordFromDateToDate.copyRecords("TABLE_NAME" , "20131101:" , "20131102:");
    }

}