示例 HBase POC

在本博客中，我们将讨论 HBase 的概念证明示例。

这里我们有一个如下图所示的数据集。

该数据集包含有关总来电、去电和特定日期从特定手机号码发送的消息的持续时间的详细信息。

第一个字段代表日期，第二个字段代表手机号码，第三个字段代表来电总时长，第四个字段代表拨出电话总时长，第五个字段代表发送的消息总数。

现在，我们的任务是从特定日期的电话号码中检索来电和去电以及发送的消息的持续时间信息。

在这个用例中，我试图过滤2014 年 3 月15^日的记录。这是一个 HBase 程序来实现这一点。

下面是它的完整代码。

public class sample {
            private static Configuration conf;
            static HTable table;
         public sample(String tableName, String colFams) throws IOException {
                        conf = HBaseConfiguration.create();
                        createTable(tableName, colFams);
                        table = new HTable(conf, tableName);
            }
            void createTable(String tableName, String colFams) throws IOException {
                        HBaseAdmin hbase = new HBaseAdmin(conf);
                        HTableDescriptor desc = new HTableDescriptor(tableName);
                        HColumnDescriptor meta = new HColumnDescriptor(colFams.getBytes());
                        desc.addFamily(meta);
                        hbase.createTable(desc);
            }
            public static void addColumnEntry(String tableName, String row,
                                    String colFamilyName, String colName, String values)
                                    throws IOException {
                        byte[] rowKey = Bytes.toBytes(row);
                        Put putdata = new Put(rowKey);
                        putdata.add(Bytes.toBytes(colFamilyName), Bytes.toBytes(colName),
                                                Bytes.toBytes(values));
                        table.put(putdata);
            }
            public static void getAllRecord(String tableName, String startPartialKey,
                                    String endPartialKey) throws IOException {
                        try {
                                    Scan s;
                                    if (startPartialKey == null || endPartialKey == null)
                                                s = new Scan();
                                    else
                                                s = new Scan(Bytes.toBytes(startPartialKey),
                                                                        Bytes.toBytes(endPartialKey));
                                    ResultScanner ss = table.getScanner(s);
                                  HashMap<String, HashMap<String, String>> outputRec = new HashMap<String, HashMap<String, String>>();
                                    String imsi = “”;
                                    for (Result r : ss) {
                                          HashMap<String, String> keyVal = new HashMap<String, String>();
                                                for (KeyValue kv : r.raw()) {
                                                      imsi = new String(kv.getRow()).substring(10);
                                                            keyVal.put(new String(kv.getQualifier()),
                                                                                 new String(kv.getValue()));
                                                            outputRec.put(imsi, keyVal);
                                                            if (keyVal.size() == 3)
                                                                 System.out.println(imsi + ” ” + “Incoming minutes:”
                                                                               + keyVal.get(“c1″) + ” Outcoming minutes:”
                                                                               + keyVal.get(“c2″) + ” Messages:”
                                                                                + keyVal.get(“c3”));
                                                }
                                    }
                        } finally {
                        }
            }
            public static void main(String[] args) throws IOException {
                        String tableName = “daterecords”;
                        String colFamilyNames = “i”;
                      sample test = new sample(tableName, colFamilyNames);
                        String fileName = “/home/cloudera/Desktop/data”;
                        // This will reference one line at a time
                        String line = null;
                        try {
                                    // FileReader reads text files in the default encoding.
                                    FileReader fileReader = new FileReader(fileName);
                                    // Always wrap FileReader in BufferedReader.
                                    BufferedReader bufferedReader = new BufferedReader(fileReader);
                                    while ((line = bufferedReader.readLine()) != null) {
                                                String[] values = line.split(” “);
                                                addColumnEntry(tableName, values[0] + “-” + values[1],
                                                                        colFamilyNames, “c1”, values[2]);
                                                addColumnEntry(tableName, values[0] + “-” + values[1],
                                                                        colFamilyNames, “c2”, values[3]);
                                                addColumnEntry(tableName, values[0] + “-” + values[1],
                                                                       colFamilyNames, “c3”, values[4]);
                                    }
                                    bufferedReader.close();
                        } catch (FileNotFoundException ex) {
                                    System.out.println(“Unable to open file ‘” + fileName + “‘”);
                        } catch (IOException ex) {
                                    System.out.println(“Error reading file ‘” + fileName + “‘”);
                                    // Or we could just do this:
                                    // ex.printStackTrace();
                        }
                        getAllRecord(tableName, “20140315”, “20140316”);
            }
}

在这里，我们创建了一个 Configuration 对象，HTable 类，并创建了名称为：daterecords和列族：i的 Hbase 表。

在此用例中，我们将使用由“-”分隔的日期和手机号码的组合作为此 Hbase 表的行键，并将传入、传出呼叫持续时间'、发送的消息数量作为列 'c1'、' c2', 'c3' 代表列族 'i'。

我们将输入数据存储在 Cloudera 的本地文件系统中。所以我们需要编写从文件中读取数据的Java Logic。

下面是Java逻辑。

在这种方法中，我们将列族的每一列的数据存储到表中。

我们可以使用 scan 命令检查存储在 Hbase 表 'daterecords' 中的数据。

您将收到如下图所示的数据。

现在我们已经成功地将数据插入到 HBase 表中。

让我们检索存储在特定日期表中的记录。

在这个用例中,我们正试图获取日期的记录：15^日2014年3月

为了检索记录，我们创建了一个方法

getAllRecord(String tableName, String startPartialKey, String endPartialKey)

第一个参数代表表名，第二个代表我们需要从中检索数据的开始日期，第三个是开始日期的下一个日期。

例如：

getAllRecord(tableName, "20140315", "20140316");

现在让我们了解这个方法的逻辑。

我们正在尝试在 startPartialKey 和 endPartialKey 的帮助下使用 HBase API 扫描 Hbase 表。

作为 StartPartialKey  和 endPartialkey 不为空 ，它将转到 else 块并扫描具有 startPartialKey 值的记录。

我们创建了一个 Result 扫描器对象，它存储 Hbase 表的扫描记录和一个 HashMap 来存储将作为结果的输出。

我们正在创建一个 Result 对象以获取 Result Scanner 中的数据存储并执行 for 循环。

imsi 是定义用于存储手机号码的字符串，keyVal 是一个哈希映射，用于存储从特定手机的列中检索到的输出。

我们已经给20140315-1234567890作为rowkey到HBase的表。其中 20140315 代表日期，1234567890 代表手机号码。

因为我们只需要手机号码，所以我们使用子字符串方法来检索它。

我们正在从 r.raw() 检索数据并使用 Put 将其存储在 HashMap 中。

最后，我们尝试在控制台上打印它们。

输出将如下图所示。

我们已成功检索到日期：2014 年 3 月15^日的记录。