Tagging contents of a SparkRDD
<p>We have data of publications and we want to tag them into various categories. We have stored them into <code>HBase</code> and from there we are saving them into <code>sparkrdd</code> and then tag them using <code>scala</code> code. A sample of the HBase data looks like this:</p>

<pre><code>PubEntity:Abstract                  timestamp=1476537886382, value=not found                                                                  
 PubEntity:Affiliations              timestamp=1476537886382, value=[]                                                                         
 PubEntity:Article_Title             timestamp=1476537886382, value=Formate assay in body fluids: application in methanol poisoning.           
 PubEntity:Author                    timestamp=1476537886382, value=[{'LastName': 'Makar', 'ForeName': 'A B', 'author_name': 'A B Makar', 'Init
                                     ials': 'AB', 'author_affiliation': 'not found'}, {'LastName': 'McMartin', 'ForeName': 'K E', 'author_name'
                                     : 'K E McMartin', 'Initials': 'KE', 'author_affiliation': 'not found'}, {'LastName': 'Palese', 'ForeName':
                                      'M', 'author_name': 'M Palese', 'Initials': 'M', 'author_affiliation': 'not found'}, {'LastName': 'Tephly
                                     ', 'ForeName': 'T R', 'author_name': 'T R Tephly', 'Initials': 'TR', 'author_affiliation': 'not found'}]  
 PubEntity:Journal_Title             timestamp=1476537886382, value=Biochemical medicine                                                       
 PubEntity:PMID                      timestamp=1476537886382, value=1                                                                          
 PubRemaining:Countries              timestamp=1476537886382, value=[]                                                                         
 PubRemaining:Created_At             timestamp=1476537886382, value=170812800.0                                                                
 PubRemaining:DOI                    timestamp=1476537886382, value=not found                                                                  
 PubRemaining:Date_Created           timestamp=1476537886382, value=19760116                                                                   
 PubRemaining:ISO_Abbreviation       timestamp=1476537886382, value=Biochem Med                                                                
 PubRemaining:ISSN                   timestamp=1476537886382, value=0006-2944                                                                  
 PubRemaining:Pub_Date               timestamp=1476537886382, value=01 Jun, 1975                                                               
 PubRemaining:Year                   timestamp=1476537886382, value=1975 
</code></pre>

<p>We have been able to make a <code>RDD</code> in spark as explained in first answer in </p>

<p><a href="https://stackoverflow.com/questions/25040709/how-to-read-from-hbase-using-spark">How to read from hbase using spark</a>
Here is the code:</p>

<pre><code>import org.apache.hadoop.hbase.client.{HBaseAdmin, Result}
import org.apache.hadoop.hbase.{ HBaseConfiguration, HTableDescriptor }
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.io.ImmutableBytesWritable

import org.apache.spark._

object HBaseRead {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("HBaseRead").setMaster("local[2]").set("spark.driver.allowMultipleContexts", "true").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(sparkConf)
    val conf = HBaseConfiguration.create()
    val tableName = "name_of_the_database"

    System.setProperty("user.name", "hdfs")
    System.setProperty("HADOOP_USER_NAME", "hdfs")
    conf.set("hbase.master", "localhost:60000")
    conf.setInt("timeout", 120000)
    conf.set("hbase.zookeeper.quorum", "localhost")
    conf.set("zookeeper.znode.parent", "/hbase-unsecure")
    conf.set(TableInputFormat.INPUT_TABLE, tableName) 

    val admin = new HBaseAdmin(conf)
    if (!admin.isTableAvailable(tableName)) {
      val tableDesc = new HTableDescriptor(tableName)
      admin.createTable(tableDesc)
    }

    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
    println(" Number of Records found : " + hBaseRDD.count())
    println(hBaseRDD.first())
    sc.stop()
  }
}
</code></pre>

<p>After running this code chunk in Scala I got: <code>defined module HBaseRead</code>
Then I did HBaseRead.main(Array()) which outputs the number of records found and reads the first record as :</p>

<p>(31,keyvalues={1/PubEntity:Abstract/1476537886382/Put/vlen=9/mvcc=0, 1/Entity:Affiliations/1476537886382/Put/vlen=2/mvcc=0, 1/Entity:Article_Title/1476537886382/Put/vlen=64/mvcc=0, 1/Entity:Author/1476537886382/Put/vlen=497/mvcc=0, 1/Entity:Journal_Title/1476537886382/Put/vlen=20/mvcc=0, 1/Entity:PMID/1476537886382/Put/vlen=1/mvcc=0, 1/Remaining:Countries/1476537886382/Put/vlen=2/mvcc=0, 1/Remaining:Created_At/1476537886382/Put/vlen=11/mvcc=0, 1/Remaining:DOI/1476537886382/Put/vlen=9/mvcc=0, 1/Remaining:Date_Created/1476537886382/Put/vlen=8/mvcc=0, 1/Remaining:ISO_Abbreviation/1476537886382/Put/vlen=11/mvcc=0, 1/Remaining:ISSN/1476537886382/Put/vlen=9/mvcc=0, 1/Remaining:Pub_Date/1476537886382/Put/vlen=12/mvcc=0, 1/Remaining:Year/1476537886382/Put/vlen=4/mvcc=0})</p>

<p>Now in this output, you would see vlen=12/mvcc=0. On checking the data, I found vlen is the length of each word/number. I couldn't figure out what mvcc is meant for. 
We want the output to show the words/number instead of vlen=4. Further we want to read these entries and find certain words and phrases in them and tag them accordingly. All in scala.</p>

<p>Any link to any helpful online resource in this regard would be highly  appreciated.</p>