如何在Spark-notebook中从Accumulo 1.6创建Spark RDD？

Question

我有Spark Notebook，Spark，Accumulo 1.6和Hadoop都在运行的Vagrant图像。从笔记本电脑，我可以手动创建一个扫描仪，并从我创建使用的Accumulo一个例子表拉力测试数据：

val instanceNameS = "accumulo" 
val zooServersS = "localhost:2181" 
val instance: Instance = new ZooKeeperInstance(instanceNameS, zooServersS) 
val connector: Connector = instance.getConnector("root", new PasswordToken("password")) 
val auths = new Authorizations("exampleVis") 
val scanner = connector.createScanner("batchtest1", auths) 

scanner.setRange(new Range("row_0000000000", "row_0000000010")) 

for(entry: Entry[Key, Value] <- scanner) { 
    println(entry.getKey + " is " + entry.getValue) 
} 

会给前十行表中的数据。

当我尝试正是如此创建RDD：

val rdd2 = 
    sparkContext.newAPIHadoopRDD (
    new Configuration(), 
    classOf[org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat], 
    classOf[org.apache.accumulo.core.data.Key], 
    classOf[org.apache.accumulo.core.data.Value] 
) 

我得到一个RDD还给我，我不能做与因以下错误：

java.io.IOException: Input info has not been set. at org.apache.accumulo.core.client.mapreduce.lib.impl.InputConfigurator.validateOptions(InputConfigurator.java:630) at org.apache.accumulo.core.client.mapreduce.AbstractInputFormat.validateOptions(AbstractInputFormat.java:343) at org.apache.accumulo.core.client.mapreduce.AbstractInputFormat.getSplits(AbstractInputFormat.java:538) at org.apache.spark.rdd.NewHadoopRDD.getPartitions(NewHadoopRDD.scala:98) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:222) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:220) at scala.Option.getOrElse(Option.scala:120) at org.apache.spark.rdd.RDD.partitions(RDD.scala:220) at org.apache.spark.SparkContext.runJob(SparkContext.scala:1367) at org.apache.spark.rdd.RDD.count(RDD.scala:927)

这完全因为我没有指定任何参数来连接哪个表，认证是什么等等这一事实是有意义的。

所以我的问题是：我该怎么做需要从这里开始将前十行表数据存入我的RDD中？

更新一个 还不行，但我确实发现了一些东西。原来，有两个几乎相同的包，

org.apache.accumulo.core.client.mapreduce

&

org.apache.accumulo.core.client.mapred

都具有几乎相同的成员，除了一个事实，即一些方法签名是不同的。不知道为什么两者都存在，因为没有我可以看到的贬低通知。我试图以不愉快的方式实现Sietse的回答。下面是我做什么，并且响应：

import org.apache.hadoop.mapred.JobConf 
import org.apache.hadoop.conf.Configuration 
val jobConf = new JobConf(new Configuration) 

import org.apache.hadoop.mapred.JobConf import org.apache.hadoop.conf.Configuration jobConf: org.apache.hadoop.mapred.JobConf = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml

Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml

AbstractInputFormat.setConnectorInfo(jobConf, 
            "root", 
            new PasswordToken("password") 

AbstractInputFormat.setScanAuthorizations(jobConf, auths) 

AbstractInputFormat.setZooKeeperInstance(jobConf, new ClientConfiguration) 

val rdd2 = 
    sparkContext.hadoopRDD (
    jobConf, 
    classOf[org.apache.accumulo.core.client.mapred.AccumuloInputFormat], 
    classOf[org.apache.accumulo.core.data.Key], 
    classOf[org.apache.accumulo.core.data.Value], 
    1 
) 

rdd2: org.apache.spark.rdd.RDD[(org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value)] = HadoopRDD[1] at hadoopRDD at :62

rdd2.first 

java.io.IOException: Input info has not been set. at org.apache.accumulo.core.client.mapreduce.lib.impl.InputConfigurator.validateOptions(InputConfigurator.java:630) at org.apache.accumulo.core.client.mapred.AbstractInputFormat.validateOptions(AbstractInputFormat.java:308) at org.apache.accumulo.core.client.mapred.AbstractInputFormat.getSplits(AbstractInputFormat.java:505) at org.apache.spark.rdd.HadoopRDD.getPartitions(HadoopRDD.scala:201) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:222) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:220) at scala.Option.getOrElse(Option.scala:120) at org.apache.spark.rdd.RDD.partitions(RDD.scala:220) at org.apache.spark.rdd.RDD.take(RDD.scala:1077) at org.apache.spark.rdd.RDD.first(RDD.scala:1110) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.(:64) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.(:69) at...

*编辑2 *

重：霍顿的答案 - 仍然没有喜悦：

AbstractInputFormat.setConnectorInfo(jobConf, 
             "root", 
             new PasswordToken("password") 
    AbstractInputFormat.setScanAuthorizations(jobConf, auths) 
    AbstractInputFormat.setZooKeeperInstance(jobConf, new ClientConfiguration) 
    InputFormatBase.setInputTableName(jobConf, "batchtest1") 
    val rddX = sparkContext.newAPIHadoopRDD(
     jobConf, 
     classOf[org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat], 
     classOf[org.apache.accumulo.core.data.Key], 
     classOf[org.apache.accumulo.core.data.Value] 
    ) 

rddX: org.apache.spark.rdd.RDD[(org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value)] = NewHadoopRDD[0] at newAPIHadoopRDD at :58

Out[15]: NewHadoopRDD[0] at newAPIHadoopRDD at :58

rddX.first 

java.io.IOException: Input info has not been set. at org.apache.accumulo.core.client.mapreduce.lib.impl.InputConfigurator.validateOptions(InputConfigurator.java:630) at org.apache.accumulo.core.client.mapreduce.AbstractInputFormat.validateOptions(AbstractInputFormat.java:343) at org.apache.accumulo.core.client.mapreduce.AbstractInputFormat.getSplits(AbstractInputFormat.java:538) at org.apache.spark.rdd.NewHadoopRDD.getPartitions(NewHadoopRDD.scala:98) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:222) at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:220) at scala.Option.getOrElse(Option.scala:120) at org.apache.spark.rdd.RDD.partitions(RDD.scala:220) at org.apache.spark.rdd.RDD.take(RDD.scala:1077) at org.apache.spark.rdd.RDD.first(RDD.scala:1110) at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.(:61) at

编辑3 - 进步！

我能弄清楚为什么'input INFO not set'错误发生。眼尖的你之间无疑会看到下面的代码缺少结束“（”

AbstractInputFormat.setConnectorInfo(jobConf, "root", new PasswordToken("password") 

为我做这个火花的笔记本电脑，我一直点击执行按钮和移动上，因为我没有看到错误。我忘记的是，当你离开关闭'''时，笔记本将做火花外壳将会做的事情 - 它会永久等待你添加它。所以错误是'setConnectorInfo'方法永远不会被执行的结果。

不幸的是，我仍然无法将accumulo表数据推送到可用于我的RDD中。当我执行

rddX.count 

我回来

res15: Long = 10000

这是正确的反应 - 有10000行中我指着表中的数据。然而，当我试图抓住正是如此数据的第一个元素：

rddX.first 

我得到以下错误：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0.0 in stage 0.0 (TID 0) had a not serializable result: org.apache.accumulo.core.data.Key

在何处何去何从有什么想法？

编辑4-成功！

接受的答案+评论是那里的90％ - 除了accumulo键/值需要被转换成可序列化的事实。通过调用两者的.toString（）方法，我得到了这个工作。我会尽快发布一些完整的工作代码，以防其他人遇到同样的问题。

Answer 1

通常使用自定义Hadoop InputFormats，信息是使用JobConf指定的。正如@Sietse指出的那样，您可以使用AccumuloInputFormat的一些静态方法来配置JobConf。在这种情况下，我认为你会想要做的是：

val jobConf = new JobConf() // Create a job conf 
// Configure the job conf with our accumulo properties 
AccumuloInputFormat.setConnectorInfo(jobConf, principal, token) 
AccumuloInputFormat.setScanAuthorizations(jobConf, authorizations) 
val clientConfig = new ClientConfiguration().withInstance(instanceName).withZkHosts(zooKeepers) 
AccumuloInputFormat.setZooKeeperInstance(jobConf, clientConfig) 
AccumuloInputFormat.setInputTableName(jobConf, tableName) 
// Create an RDD using the jobConf 
val rdd2 = sc.newAPIHadoopRDD(jobConf, 
classOf[org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat], 
classOf[org.apache.accumulo.core.data.Key], 
classOf[org.apache.accumulo.core.data.Value] 
) 

注：挖掘到代码后，似乎在配置属性基于在被称为（类部分设置有道理避免与其他软件包发生冲突），所以当我们稍后返回到具体类时，它无法找到配置标志。解决方案是不使用抽象类。有关实施细节，请参阅https://github.com/apache/accumulo/blob/bf102d0711103e903afa0589500f5796ad51c366/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/ConfiguratorBase.java#L127）。如果你不能用spark-notebook的具体实现来调用这个方法，可能使用spark-shell或者定期构建的应用程序是最简单的解决方案。

Answer 2

看起来这些参数必须通过静态方法设置：http://accumulo.apache.org/1.6/apidocs/org/apache/accumulo/core/client/mapred/AccumuloInputFormat.html。因此请尝试设置非可选参数并再次运行。它应该工作。