使用星火

Question

我使用下面的代碼試圖將數據存儲在HBase的在HBase的存儲數據：

val x = Seq((1, ("a", "A")), (1, ("a", "AA")), (2, ("d", "D")), (2, ("d", "DD"))) 
val f = sc.parallelize(x) 
val z = f.groupByKey() 

z.collectAsMap().foreach(elem => { 
    var rowKey = elem._1.toString() 
    var p = new Put(rowKey.getBytes()) 

    elem._2.foreach(innerElem => { 
     var col = innerElem._1 
     var value = innerElem._2 
     p.add("cf".getBytes(), new String(col).getBytes(), new String(value).getBytes()) 
     table.put(p) 
    }) 
}) 
table.flushCommits() 

我正在從HBase的外殼下面的輸出：

ROW  COLUMN+CELL                     
1   column=cf:a, timestamp=14879172, value=AA                         
2   column=cf:d, timestamp=1487917201226, value=DD             

但我想得到：

ROW  COLUMN+CELL 
1   column=cf:a, timestamp=14879172, value=A                     
1   column=cf:a, timestamp=148791720123X, value=AA             
2   column=cf:d, timestamp=1487917201226, value=D             
2   column=cf:d, timestamp=148791720122X, value=DD 

我的代碼覆蓋特定列的第一個值，並只存儲該列的最後一個值。我想存儲特定列的每個值。

當我移動到var p = new Put(rowKey.getBytes())斯卡拉殼第二foreach循環我有以下幾點：

scala>  
    | z.collectAsMap().foreach(elem => { 
    |  val rowKey = elem._1.toString() 
    |   
    |  elem._2.foreach(innerElem => { 
    | Display all 620 possibilities? (y or n) 
    | ew Put(rowKey.getBytes()) 
    |   var col = innerElem._1 
    |   var value = innerElem._2 
    |   p.add("cf".getBytes(), new String(col).getBytes(), new String(value).getBytes()) 
    |   table.put(p) 
    |  }) 
    |  }) 
<console>:49: error: not found: value ew 
     ew Put(rowKey.getBytes()) 
    ^
<console>:52: error: not found: value p 
       p.add("cf".getBytes(), new String(col).getBytes(), new String(value).getBytes()) 
      ^
<console>:53: error: not found: value p 
       table.put(p) 
         ^

Answer 1

你的問題就出在Put對象的範圍。把它想成一個哈希表。您只需重寫第二個迭代中的值即可。在每次迭代中創建單獨的Put對象，它會很好。