java - 从 Storm bolt 将行插入 HBase

Question

我希望能够从分布式(非本地)Storm 拓扑中将新条目写入 HBase。有一些 GitHub 项目提供 HBase Mappers或 pre-made Storm bolts将元组写入 HBase。这些项目提供了在 LocalCluster 上执行示例的说明。

我在这两个项目中遇到的问题，以及直接从 bolt 访问 HBase API 的问题是，它们都需要将 HBase-site.xml 文件包含在类路径中。使用直接 API 方法，也许还有 GitHub 方法，当您执行 HBaseConfiguration.create(); 时，它将尝试从类路径上的条目中查找所需的信息。

如何修改 storm bolts 的类路径以包含 Hbase 配置文件？

更新:使用 danehammer 的回答，这就是我让它工作的方式

将以下文件复制到您的 ~/.storm 目录中:

• hbase-common-0.98.0.2.1.2.0-402-hadoop2.jar
• hbase-site.xml
• storm.yaml :注意:如果您不将 storm.yaml 复制到该目录，则 storm jar 命令将不会在类路径中使用该目录(请参阅 storm.py python script 自己查看该逻辑 - 会很好如果记录在案)

接下来，在拓扑类的主要方法中获取 HBase 配置并将其序列化:

final Configuration hbaseConfig = HBaseConfiguration.create();
final DataOutputBuffer databufHbaseConfig = new DataOutputBuffer();
hbaseConfig.write(databufHbaseConfig);
final byte[] baHbaseConfigSerialized = databufHbaseConfig.getData();

通过构造函数将字节数组传递给您的 spout 类。 spout 类将这个字节数组保存到一个字段(不要在构造函数中反序列化。我发现如果 spout 有一个 Configuration 字段，你会在运行拓扑时得到一个无法序列化的异常)

在spout的open方法中，反序列化配置并访问hbase表:

Configuration hBaseConfiguration = new Configuration();
ByteArrayInputStream bas = new ByteArrayInputStream(baHbaseConfigSerialized);
hBaseConfiguration.readFields(new DataInputStream(bas));
HTable tbl = new HTable(hBaseConfiguration, HBASE_TABLE_NAME);

Scan scan = new Scan();
scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("YOUR_COLUMN"));

scnrTbl = tbl.getScanner(scan);

现在，在您的 nextTuple 方法中，您可以使用 Scanner 获取下一行:

Result rsltWaveform = scnrWaveformTbl.next();

从结果中提取您想要的内容，并将某些可序列化对象中的这些值传递给 bolt 。

Answer 1

当您使用“storm jar”命令部署拓扑时，~/.storm 文件夹将位于类路径中(参见 jar 命令下的 this link)。如果您将 hbase-site.xml 文件(或相关的 *-site.xml 文件)放在该文件夹中，HBaseConfiguration.create() 在“storm jar”期间会找到该文件并正确返回给您org.apache.hadoop.configuration.Configuration。这需要在您的拓扑中存储和序列化，以便在集群中分发该配置。