HBaseを使って簡易ブログサービスを作ってみよう

初めてのHBase　第3回

ブログ記事の取得

　次に、ブログ記事の取得のコードになります。以下は、最新順で取得するコードです。

// ブログ記事の取得(最新順)
public List<Article> getArticles(long userId, int length, Article lastArticle) throws IOException {
  // Scanオブジェクト
  Scan scan = new Scan();
  if (lastArticle == null) {
    scan.setStartRow(createStartRow(userId));
  } else {
    // ページング処理
    scan.setStartRow(createPagingStartRow(userId, lastArticle.getPostAt(), lastArticle.getArticleId()));
  }
  scan.setStopRow(createStopRow(userId));
  
  List<Article> ret = new ArrayList<Article>();
  
  // プールからHTableInterfaceを取得
  HTableInterface table = hTablePool.getTable(TABLE);
  ResultScanner scanner = null;
  try {
    // ResultScannerの取得
    scanner = table.getScanner(scan);
    
    for (Result result : scanner) {
      // データを取得
      byte[] value = result.getValue(COLUMN_FAMILY, HConstants.EMPTY_BYTE_ARRAY);
      // データのデシリアライズ
      Article article = deserialize(value);
      ret.add(article);
      if (ret.size() >= length) {
        break;
      }
    }
  } finally {
    if (scanner != null) {
      scanner.close();
    }
    table.close();
  }
  return ret;
}

// startRowの作成。hash(userId)-userId-0
private byte[] createStartRow(long userId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1); // int型 + long型 + byte型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) 0); // 0
  return buffer.array();
}

// stopRowの作成。hash(userId)-userId-(0 + 1)
private byte[] createStopRow(long userId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1); // int型 + long型 + byte型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) (0 + 1)); // 0 + 1．パーシャルScan
  return buffer.array();
}

// ページング用startRowの作成。hash(userId)-userId-0-(Long.MAX_VALUE - lastPostAt)-(lastArticleId + 1)
private byte[] createPagingStartRow(long userId, long lastPostAt, long lastArticleId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1 + 8 + 8); // int型 + long型 + byte型 + long型 + long型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) 0); // 0
  buffer.putLong(Long.MAX_VALUE - lastPostAt); // Long.MAX_VALUE - lastPostAt
  buffer.putLong(lastArticleId + 1); // articleId + 1．最後の記事は含めないので+1する
  return buffer.array();
}

// byte[]からArticleにデシリアライズする
private Article deserialize(byte[] bytes) {
  // 省略
}

　Scanオブジェクトを生成し、startRowとstopRowを指定してResultScannerを取得し、結果をデシリアライズして返しています。ここでのポイントは、startRowとstopRowをどのように指定するかということです。

　ここでパーシャルScanというテクニックを紹介します。Scan時にstartRowとstopRowを指定するとき、本来はそれぞれを正確に指定したいところですが、いつもstopRowが分かるわけではありません。今回の簡易ブログサービスでも、正確なstopRowを指定するためには、ユーザが最初に投稿した記事のpostAtとarticleIdを覚えておかなければなりません。

　そこで、stopRowにstartRow+1を指定してあげるのがパーシャルScanです。今回の簡易ブログサービスでは、startRowにhash(userId)-userId-0を指定し、stopRowにhash(userId)-userId-(0+1)を指定しています。このようにすることで、複雑なことをせずにScanを実行できます。

　次に、ページングについてです。勘の良い読者であればもうお分かりでしょう。lastArticleを引数として受け取っているので、そのpostAtとarticleIdを用いてhash(userId)-userId-0-(Long.MAX_VALUE-lastPostAt)-(lastArticleId+1)のRowKeyを作成しstartRowに指定します。lastArticleIdに1を足しているのは、最後の記事を含めないためです。

　このようにして、前の結果の次の記事からScanを始めることができるので、高速なページングを実現することができます。

　以下は、カテゴリ別で取得するコードです。基本的に、最新順で取得するコードと同じです。違いはセカンダリインデックス用のスキーマに対して、Scanを行っていることです。

// ブログ記事の取得(カテゴリ別)
public List<Article> getArticles(long userId, int categoryId, int length, Article lastArticle) throws IOException {
  // Scanオブジェクト
  Scan scan = new Scan();
  
  if (lastArticle == null) {
    scan.setStartRow(createSecondaryIndexStartRow(userId, categoryId));
  } else {
    // ページング処理
    scan.setStartRow(createSecondaryIndexPagingStartRow(userId, categoryId, lastArticle.getPostAt(), lastArticle.getArticleId()));
  }
  scan.setStopRow(createSecondaryIndexStopRow(userId, categoryId));
  
  List<Article> ret = new ArrayList<Article>();
  
  // プールからHTableInterfaceを取得
  HTableInterface table = hTablePool.getTable(TABLE);
  ResultScanner scanner = null;
  try {
    scanner = table.getScanner(scan);
    
    for (Result result : scanner) {
      byte[] value = result.getValue(COLUMN_FAMILY, HConstants.EMPTY_BYTE_ARRAY);
      Article article = deserialize(value);
      ret.add(article);
      if (ret.size() >= length) {
        break;
      }
    }
  } finally {
    if (scanner != null) {
      scanner.close();
    }
    table.close();
  }
  return ret;
}

// セカンダリインデックスのstartRowの作成。hash(userId)-userId-1-categoryId
private byte[] createSecondaryIndexStartRow(long userId, int categoryId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1 + 4); // int型 + long型 + byte型 + int型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) 1); // 1
  buffer.putInt(categoryId); // categoryId
  return buffer.array();
}

// セカンダリインデックスのstopRowの作成。hash(userId)-userId-1-(categoryId + 1)
private byte[] createSecondaryIndexStopRow(long userId, int categoryId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1 + 4); // int型 + long型 + byte型 + int型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) 1); // 1
  buffer.putInt(categoryId + 1); // categoryId + 1．パーシャルScan
  return buffer.array();
}

// セカンダリインデックスのページング用startRowの作成。hash(userId)-userId-1-categoryId-(Long.MAX_VALUE - lastPostAt)-(lastArticleId + 1)
private byte[] createSecondaryIndexPagingStartRow(long userId, int categoryId, long lastPostAt, long lastArticleId) {
  ByteBuffer buffer = ByteBuffer.allocate(4 + 8 + 1 + 4 + 8 + 8); // int型 + long型 + byte型 + int型 + long型 + long型
  buffer.putInt(hash(userId)); // hash(userId)
  buffer.putLong(userId); // userId
  buffer.put((byte) 1); // 1
  buffer.putInt(categoryId); // categoryId
  buffer.putLong(Long.MAX_VALUE - lastPostAt); // Long.MAX_VALUE - lastPostAt
  buffer.putLong(lastArticleId + 1); // articleId + 1．最後の記事は含めないので+1する
  return buffer.array();
}

まとめ

　今回は、HBaseを使って簡易ブログサービスを作りました。実際にアプリケーションを作ることで、HBaseのスキーマ設計についてイメージがついたのではないでしょうか。次回は、HBaseを使ったアプリケーションの別のケーススタディを紹介したいと思います。

　また、株式会社サイバーエージェントでは、Hadoop/HBaseエンジニアを募集しています。ご興味のある方はこちらからエントリーしていただければと思います。エンジニア>R&Dエンジニア>R&Dエンジニアを選択しエントリーしていただければ幸いです。

参考資料

• Apache HBase Home
• 『HBase』