如何在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。因此請嘗試設置非可選參數並再次運行。它應該工作。