Hbase - 自定義Rowkey規(guī)則

> 在Flink中我們有時候需要分析數(shù)據(jù)1點(diǎn)到2點(diǎn)的范圍,可是經(jīng)過Region又比較慢，這時候我們就可以定制`TableInputFormat`來實(shí)現(xiàn)我們的需求了，我們還可以采用Flink的`DataSet`的方式讀取,另外下面還有`Spark`讀取的例子。

## 使用教程

Md5Util.java

```

import org.apache.commons.codec.binary.Hex;

import java.security.MessageDigest;

import java.security.NoSuchAlgorithmException;

public class Md5Util {

? ? public static String md5(byte[] key) {

? ? ? ? return md5(key, 0, key.length);

? ? }

? ? public static String md5(byte[] key, int offset, int length) {

? ? ? ? try {

? ? ? ? ? ? MessageDigest e = MessageDigest.getInstance("MD5");

? ? ? ? ? ? e.update(key, offset, length);

? ? ? ? ? ? byte[] digest = e.digest();

? ? ? ? ? ? return new String(Hex.encodeHex(digest));

? ? ? ? } catch (NoSuchAlgorithmException var5) {

? ? ? ? ? ? throw new RuntimeException("Error computing MD5 hash", var5);

? ? ? ? }

? ? }

? ? public static String md5(String str) {

? ? ? ? return md5(str.getBytes());

? ? }

? ? public static String md5(String str,int offset, int length) {

? ? ? ? return md5(str.getBytes(),offset,length);

? ? }

}

```

數(shù)據(jù)`Split`方式

```

private Connection connection;

? ? private Admin admin;

? ? @Before

? ? public void init() throws Exception {

? ? ? ? System.setProperty("java.security.krb5.conf", "/etc/krb5.conf");

? ? ? ? System.setProperty("sun.security.krb5.debug", "false");

? ? ? ? final String user = "hbase/abc.demo.com@DEMO.COM";

? ? ? ? final String keyPath = "/home/dounine/kerberos/lake.keytab";

? ? ? ? Configuration conf = new Configuration();

? ? ? ? conf.addResource("hbase-site.xml");

? ? ? ? UserGroupInformation.setConfiguration(conf);

? ? ? ? UserGroupInformation.loginUserFromKeytab(user, keyPath);

? ? ? ? connection = ConnectionFactory.createConnection(conf);

? ? ? ? admin = connection.getAdmin();

? ? }

@Test

? ? public void createTable() throws IOException {

? ? ? ? TableName table = TableName.valueOf("logTable1");

? ? ? ? TableDescriptorBuilder tableDesc = TableDescriptorBuilder.newBuilder(table);

? ? ? ? tableDesc.setValue(TableDescriptorBuilder.SPLIT_POLICY,KeyPrefixRegionSplitPolicy.class.getName());

? ? ? ? tableDesc.setValue(KeyPrefixRegionSplitPolicy.PREFIX_LENGTH_KEY,"2");

? ? ? ? ColumnFamilyDescriptor extCF = ColumnFamilyDescriptorBuilder.newBuilder("ext".getBytes()).build();

? ? ? ? ColumnFamilyDescriptor deviceCF = ColumnFamilyDescriptorBuilder.newBuilder("device".getBytes()).build();

? ? ? ? ColumnFamilyDescriptor locationCF = ColumnFamilyDescriptorBuilder.newBuilder("location".getBytes()).build();

? ? ? ? tableDesc.setColumnFamilies(Arrays.asList(extCF,locationCF,deviceCF));

? ? ? ? try {

? ? ? ? ? ? byte[][] splitKeys = new byte[4][];

? ? ? ? ? ? splitKeys[0] = Bytes.toBytes("00");

? ? ? ? ? ? splitKeys[1] = Bytes.toBytes("40");

? ? ? ? ? ? splitKeys[2] = Bytes.toBytes("80");

? ? ? ? ? ? splitKeys[3] = Bytes.toBytes("c0");

? ? ? ? ? ? admin.createTable(tableDesc.build(),splitKeys);

? ? ? ? } catch (IOException e) {

? ? ? ? ? ? e.printStackTrace();

? ? ? ? }

? ? }

```

`logTable1`數(shù)據(jù)寫入方式

```

public class HbaseKerberos{

? ? private static final Logger LOGGER = LoggerFactory.getLogger(HbaseKerberos.class);

? ? private static final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS");

? ? private static final String TABLE_NAME = "logTable1";

? ? public void insertDataToHbase1(String appKey,List<Log> hasDatas) throws IOException {

? ? ? ? Table table = HbaseUtils.getTable(TABLE_NAME);

? ? ? ? Long sumCount = 0L;

? ? ? ? /**

? ? ? ? * 常規(guī)值

? ? ? ? */

? ? ? ? byte[] extCF = Bytes.toBytes("ext");//CF列族

? ? ? ? Random random = new Random();

? ? ? ? List<Put> rows = new ArrayList<>();

? ? ? ? for (Log logEntity : hasDatas) {

? ? ? ? ? ? JSONObject dataJsonObject = logEntity.getData();

? ? ? ? ? ? JSONObject extJsonObject = dataJsonObject.getJSONObject("ext");

? ? ? ? ? ? String userId = extJsonObject.getString("userId");

? ? ? ? ? ? String timeStr = logEntity.getTime().format(dtf);

? ? ? ? ? ? String md5Str = Md5Util.md5(userId);

? ? ? ? ? ? String rowKey = new StringBuilder()

? ? ? ? ? ? ? ? ? ? .append(md5Str.substring(0,2))//md5出來的前兩位最高為ff,00~ff為256位蝇狼，后期Region可以增加那么多，足夠使用了。

? ? ? ? ? ? ? ? ? ? .append("|")

? ? ? ? ? ? ? ? ? ? .append(timeStr)//時間

? ? ? ? ? ? ? ? ? ? .append("|")

? ? ? ? ? ? ? ? ? ? .append(CrcUtil.getCrcValue(appKey))

? ? ? ? ? ? ? ? ? ? .append("|")

? ? ? ? ? ? ? ? ? ? .append(md5Str.substring(2,8))

? ? ? ? ? ? ? ? ? ? .append("|")

? ? ? ? ? ? ? ? ? ? .append(Md5Util.md5(UUID.randomUUID().toString()).substring(0,2))

? ? ? ? ? ? ? ? ? ? .toString();

? ? ? ? ? ? Put row = new Put(Bytes.toBytes(rowKey));

? ? ? ? ? ? for(String keyName : extJsonObject.keySet()){

? ? ? ? ? ? ? ? String value = extJsonObject.getString(keyName);

? ? ? ? ? ? ? ? if(StringUtils.isNotBlank(value)){

? ? ? ? ? ? ? ? ? ? row.addColumn(extCF, Bytes.toBytes(keyName), Bytes.toBytes(value));

? ? ? ? ? ? ? ? }

? ? ? ? ? ? }

? ? ? ? ? ? row.addColumn(extCF, Bytes.toBytes("time"), Bytes.toBytes(logEntity.getTime().toString()));

? ? ? ? ? ? /**

? ? ? ? ? ? * 設(shè)備信息

? ? ? ? ? ? */

? ? ? ? ? ? putFieldToRow(logEntity.getData(),"device",row);

? ? ? ? ? ? /**

? ? ? ? ? ? * 位置信息

? ? ? ? ? ? */

? ? ? ? ? ? putFieldToRow(logEntity.getData(),"location",row);

? ? ? ? ? ? rows.add(row);

? ? ? ? }

? ? ? ? for(Integer[] durtation : LimitUtil.getLimits(rows.size(),1000)){

? ? ? ? ? ? Object[] results = new Object[(durtation[1]-durtation[0])];

? ? ? ? ? ? try {

? ? ? ? ? ? ? ? table.batch(rows.subList(durtation[0], durtation[1]),results);

? ? ? ? ? ? } catch (InterruptedException e) {

? ? ? ? ? ? ? ? e.printStackTrace();

? ? ? ? ? ? }

? ? ? ? ? ? sumCount += (durtation[1]-durtation[0]);

? ? ? ? }

? ? ? ? LOGGER.info("write data count:" + sumCount);

? ? }

}

```

`logTable1`數(shù)據(jù)

```

00|20180518203401772|2352356512|4519 column=ext:appKey, timestamp=1533646292389, value=898b7e90-5754-11e8-983c-6b4bcc3b7c2e? ? ? ? ? ? ? ? ? ? ?