批处理作业(Spark)，其查找表太大，无法装入内存

Question

我正在尝试编写一个批处理作业，以处理目前位于HBase数据库(AWS中的EMR集群中)中的数百兆字节，所有这些数据都位于一个大表中。对于我正在处理的每一行，我需要从第二个HBase表中的查找表(一个简单的整数到字符串映射)获得额外的数据。我们每排要查5-10次。

我的当前实现使用了一个Spark作业，该作业将输入表的分区分发给它的工作人员，其形状如下：

Configuration hBaseConfig = newHBaseConfig();
hBaseConfig.set(TableInputFormat.SCAN, convertScanToString(scan));
hBaseConfig.set(TableInputFormat.INPUT_TABLE, tableName);

JavaPairRDD<ImmutableBytesWritable, Result> table = sparkContext.newAPIHadoopRDD(hBaseConfig, TableInputFormat.class, ImmutableBytesWritable.class, Result.class);
table.map(val -> { 
    // some preprocessing  
}).foreachPartition(p -> {
    p.forEachRemaining(row -> {
        // code that does the lookup
    });
});

问题是查找表太大，无法容纳工人的记忆。它们都需要访问查找表的所有部分，但它们的访问模式将从缓存中明显受益。

我认为不能使用简单的映射作为广播变量，因为它需要放入内存，这是否正确？

Spark使用共享的nothing体系结构，所以我想在所有工作人员之间共享缓存并不是一种简单的方法，但是我们能为每个工作人员构建一个简单的LRU缓存吗？

如何实现这样一个本地工作缓存，在缓存丢失时从HBase中的查找表中获取数据？我能以某种方式将第二个表的引用分发给所有员工吗？

除了HBase作为数据源之外，我对我的技术选择不感兴趣。是否有一个比星火更适合我的用例的框架？

Answer 1

您有几个处理此要求的选项：

1-使用RDD或数据集连接

您可以将两个HBase表加载为Spark或数据集，然后在查找键上执行join。Spark将将两个RDD分割成分区和周围的洗牌内容，这样具有相同键的行就会在相同的执行器上结束。通过管理spark中的分区数量，您应该能够在任意大小上连接2个表。

2-广播解析器实例

您可以广播一个解析器实例来执行HBase查找和临时LRU缓存，而不是广播一个映射。每个执行器将获得此实例的副本，并可以管理自己的缓存，您可以在其中调用它们以获取foreachPartition()代码。

请注意，解析器实例需要实现可序列化，因此您必须将缓存、HBase连接和HBase配置属性声明为瞬态，以便在每个执行器上初始化。

我在Scala中对我维护的一个项目运行了这样一个设置:如果您知道您的访问模式并有效地管理您的缓存，它的工作效率和效率都会比直接星火连接更高效。

3-使用HBase火花连接器实现查找逻辑

Apache HBase最近加入了改进的HBase火花连接器文档，现在文档非常稀少，您需要查看JIRA票据和这些工具的前一个版本的文档( Cloudera氏SparkOnHBase )，但是测试套件中的最后一个单元测试看起来很像您想要的。

不过，我对这个API没有经验。