[Put集合] #flink #hbase

PutCollection

import com.google.common.collect.Maps;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

/**
 * 各种Put的集合，包含表信息
 */
public class PutCollection {


    public HashMap<String, Map<String, SealedPut>> getData() {
        return data;
    }

    public void setData(HashMap<String, Map<String, SealedPut>> data) {
        this.data = data;
    }

    private HashMap<String, Map<String, SealedPut>> data = Maps.newHashMap();

    public void putAll(String tableName, Collection<SealedPut> puts) {
        if (!data.containsKey(tableName)) {
            data.put(tableName, Maps.newHashMap());
        }
        final Map<String, SealedPut> table = data.get(tableName);
        for (SealedPut put : puts) {
            final String hexRow = Bytes.toHex(put.getRowKey());
            if (!table.containsKey(hexRow)) {
                table.put(hexRow, put);
            } else {
                table.put(
                        hexRow,
                        table.get(hexRow).appendSealedPut(put)
                );
            }
        }
    }

    public void put(String tableName, SealedPut put) {
        if (!data.containsKey(tableName)) {
            data.put(tableName, Maps.newHashMap());
        }
        final Map<String, SealedPut> table = data.get(tableName);
        final String hexRow = Bytes.toHex(put.getRowKey());
        if (!table.containsKey(hexRow)) {
            table.put(hexRow, put);
        } else {
            table.put(
                    hexRow,
                    table.get(hexRow).appendSealedPut(put)
            );
        }
    }


    /**
     * 合并多个PutCollection
     *
     * @param cs
     * @return
     */
    public static PutCollection merge(PutCollection... cs) {
        PutCollection c0 = new PutCollection();
        for (PutCollection ci : cs) {
            for (Map.Entry<String, Map<String, SealedPut>> entry : ci.data.entrySet()) {
                c0.putAll(entry.getKey(), entry.getValue().values());
            }
        }
        return c0;
    }

    public static PutCollection create(String tableName, Collection<SealedPut> puts) {
        PutCollection result = new PutCollection();
        result.putAll(tableName, puts);
        return result;
    }

    public static PutCollection create(String tableName, SealedPut put) {
        PutCollection result = new PutCollection();
        result.put(tableName, put);
        return result;
    }
}