title: Dependency Issues for Cloud DataProc + Spark + Cloud BigTable with JAVA
body: <p>I need to create an application to run on Cloud DataProc and process large BigTable writes, scans, and deletes in massively parallel fashion using Spark. This could be in JAVA (or Python if it's doable).</p>
  
  <p>I am trying to write the minimum code using Eclipse to achieve the basic functionality of getting an <code>RDD</code> out of a BigTable table, either by using <code>bulkPut</code>/<code>bulkDelete</code>/<code>butkGet</code> or by using <code>newAPIHadoopRDD()</code> or something similar.</p>
  
  <p>I have seen multiple posts on SO and elsewhere how this can be done and on various challenges with connecting Bigtable API, HBase API, and Spark. Some of these posts are quite outdated by now (couple years old, so may not be relevant). So far I haven't managed to get anything working, mostly due to various dependency clashes or inconsistencies. No matter what combination of dependencies and versions I try in POM.XML I get either ClassNotFound or NoSuchMethod exception when I try running things.</p>
  
  <p>Can I get some advice please on what are the 'working' combinations of Spark, HBase, and Bigtable dependency versions and packages that I need to include? My POM.xml currently looks like below. </p>
  
  <pre><code>&lt;project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"&gt;
    &lt;modelVersion&gt;4.0.0&lt;/modelVersion&gt;
    &lt;groupId&gt;FFSpark5&lt;/groupId&gt;
    &lt;artifactId&gt;FFSpark5&lt;/artifactId&gt;
    &lt;version&gt;0.0.1-SNAPSHOT&lt;/version&gt;
  
  
      &lt;properties&gt;
          &lt;bigtable.version&gt;1.0.0&lt;/bigtable.version&gt;
          &lt;hbase.version&gt;1.3.1&lt;/hbase.version&gt;
          &lt;hbase-shaded.version&gt;2.0.0-alpha2&lt;/hbase-shaded.version&gt;
          &lt;hbase-spark.version&gt;2.0.0-alpha4&lt;/hbase-spark.version&gt;
          &lt;maven.compiler.target&gt;1.8&lt;/maven.compiler.target&gt;
          &lt;maven.compiler.source&gt;1.8&lt;/maven.compiler.source&gt;
          &lt;spark.version&gt;1.6.2&lt;/spark.version&gt;
          &lt;spark-streaming.version&gt;1.6.2&lt;/spark-streaming.version&gt;
          &lt;scala.version&gt;2.11.0&lt;/scala.version&gt;
          &lt;scalatest.version&gt;3.0.3&lt;/scalatest.version&gt;
          &lt;bigtable.projectID&gt;my_gcp_project_id&lt;/bigtable.projectID&gt;
          &lt;bigtable.instanceID&gt;my_bigtable_instance_name&lt;/bigtable.instanceID&gt;
      &lt;/properties&gt;
  
      &lt;dependencies&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
              &lt;artifactId&gt;spark-core_2.11&lt;/artifactId&gt;
              &lt;version&gt;${spark.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
              &lt;artifactId&gt;spark-streaming_2.11&lt;/artifactId&gt;
              &lt;version&gt;${spark-streaming.version}&lt;/version&gt;
              &lt;scope&gt;provided&lt;/scope&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
            &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
            &lt;artifactId&gt;hbase-spark&lt;/artifactId&gt;
           &lt;version&gt;${hbase-spark.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;com.google.cloud.bigtable&lt;/groupId&gt;
              &lt;artifactId&gt;bigtable-hbase-1.x-hadoop&lt;/artifactId&gt;
              &lt;version&gt;${bigtable.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
              &lt;artifactId&gt;hbase-server&lt;/artifactId&gt;
              &lt;version&gt;${hbase.version}&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
              &lt;artifactId&gt;hbase-client&lt;/artifactId&gt;
              &lt;version&gt;${hbase.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.scala-lang&lt;/groupId&gt;
              &lt;artifactId&gt;scala-library&lt;/artifactId&gt;
              &lt;version&gt;${scala.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
              &lt;artifactId&gt;hbase-shaded-server&lt;/artifactId&gt;
              &lt;version&gt;${hbase.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
              &lt;artifactId&gt;hbase-shaded-client&lt;/artifactId&gt;
              &lt;version&gt;${hbase-shaded.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.hbase&lt;/groupId&gt;
              &lt;artifactId&gt;hbase-shaded-server&lt;/artifactId&gt;
              &lt;version&gt;${hbase-shaded.version}&lt;/version&gt;
          &lt;/dependency&gt;
  
      &lt;/dependencies&gt;
  
      &lt;build&gt;
      &lt;outputDirectory&gt;target/classes&lt;/outputDirectory&gt;
  
      &lt;resources&gt;
          &lt;resource&gt;
              &lt;directory&gt;src/main/resources&lt;/directory&gt;
              &lt;filtering&gt;true&lt;/filtering&gt;
          &lt;/resource&gt;
      &lt;/resources&gt;
  
      &lt;plugins&gt;
        &lt;plugin&gt;
          &lt;artifactId&gt;maven-assembly-plugin&lt;/artifactId&gt;
          &lt;executions&gt;
            &lt;execution&gt;
              &lt;phase&gt;package&lt;/phase&gt;
              &lt;goals&gt;
                  &lt;goal&gt;single&lt;/goal&gt;
              &lt;/goals&gt;
            &lt;/execution&gt;
          &lt;/executions&gt;
          &lt;configuration&gt;
            &lt;descriptorRefs&gt;
                &lt;descriptorRef&gt;jar-with-dependencies&lt;/descriptorRef&gt;
            &lt;/descriptorRefs&gt;
            &lt;archive&gt;
              &lt;manifest&gt;
                  &lt;mainClass&gt;FFSpark5&lt;/mainClass&gt;
              &lt;/manifest&gt;
            &lt;/archive&gt;
          &lt;/configuration&gt;
        &lt;/plugin&gt;
      &lt;/plugins&gt;
    &lt;/build&gt;
  &lt;/project&gt;
  </code></pre>
  
  <p>I realize there is probably a lot of wrong stuff in this version of POM.xml, but I have played around with many combinations of dependencies and versions, and couldn't get any of them do actually work. This latest one seems to be going the furthest in terms of the log outputs, but still breaks. Here's the latest stack trace:</p>
  
  <pre><code>18/03/12 15:37:17 INFO BigtableSession: Bigtable options: BigtableOptions{dataHost=bigtable.googleapis.com, tableAdminHost=bigtableadmin.googleapis.com, instanceAdminHost=bigtableadmin.googleapis.com .... (lost of other options here)}.
  18/03/12 15:37:17 INFO RefreshingOAuth2CredentialsInterceptor: Refreshing the OAuth token
  18/03/12 15:37:19 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 210.6 KB, free 210.6 KB)
  18/03/12 15:37:19 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 19.9 KB, free 230.5 KB)
  18/03/12 15:37:19 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:58982 (size: 19.9 KB, free: 457.9 MB)
  18/03/12 15:37:19 INFO SparkContext: Created broadcast 0 from broadcast at HBaseContext.scala:73
  18/03/12 15:37:19 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 160.0 B, free 230.6 KB)
  18/03/12 15:37:19 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 120.0 B, free 230.7 KB)
  18/03/12 15:37:19 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on localhost:58982 (size: 120.0 B, free: 457.9 MB)
  18/03/12 15:37:19 INFO SparkContext: Created broadcast 1 from broadcast at HBaseContext.scala:74
  Direct test done
  Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.hbase.client.Scan.getMvccReadPoint()J
      at org.apache.hadoop.hbase.client.PackagePrivateFieldAccessor.getMvccReadPoint(PackagePrivateFieldAccessor.java:39)
      at org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil.toScan(ProtobufUtil.java:1088)
      at org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil.convertScanToString(TableMapReduceUtil.java:601)
      at FFSpark5.main(FFSpark5.java:64)
  </code></pre>
  
  <p>And below is my basic code. The idea is to do 3 tests:</p>
  
  <p>Test 1: simply tries to access Bigtable directly via the simple Bigtable API, just to make sure there are no simple issues, like authentication, etc. This works fine</p>
  
  <p>Test 2: tries to get newAPIHadoopRDD(). This fails</p>
  
  <p>Test 3: tries to get bulkPut(). This fails</p>
  
  <pre><code>import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hbase.HBaseConfiguration;
  import org.apache.hadoop.hbase.TableName;
  import org.apache.spark.SparkConf;
  import org.apache.spark.api.java.JavaPairRDD;
  import org.apache.spark.api.java.JavaRDD;
  import org.apache.spark.api.java.JavaSparkContext;
  import org.apache.spark.api.java.function.Function;
  import org.apache.hadoop.hbase.client.ConnectionFactory;
  import org.apache.hadoop.hbase.client.Put;
  import org.apache.hadoop.hbase.client.Result;
  import org.apache.hadoop.hbase.client.Scan;
  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
  import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
  import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
  import org.apache.hadoop.hbase.spark.JavaHBaseContext;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.hadoop.hbase.client.Connection;
  
  class PutFunction implements Function&lt;String, Put&gt; {
  
      private static final long serialVersionUID = 1L;
  
  
      public Put call(String v) throws Exception {
        String[] cells = v.split(",");
        Put put = new Put(Bytes.toBytes(cells[0]));
  
        put.addColumn(Bytes.toBytes(cells[1]), Bytes.toBytes(cells[2]),
                Bytes.toBytes(cells[3]));
        return put;
      }
    }
  
  
  public class FFSpark5
  {
      public static void main(String args[]) throws IOException
      {
          SparkConf conf = new SparkConf().setAppName("SparkTest").setMaster("local");        
          JavaSparkContext sc = new JavaSparkContext(conf);       
          Configuration hBaseConf = HBaseConfiguration.create();
          Connection connection = ConnectionFactory.createConnection(hBaseConf);
          JavaHBaseContext hbaseContext = new JavaHBaseContext(sc, hBaseConf);
  
          // test 1: simple direct Bigtable access
          connection.getTable(TableName.valueOf("FFTransFlat".getBytes()))
              .put(new Put("abc".getBytes())
              .addColumn("CI".getBytes(), "I".getBytes(), "val".getBytes()));
          System.out.println("Direct test done");
  
  
          // Test 2: newAPIHadoopRDD() 
          Scan scan1 = new Scan();
          scan1.setCaching(500);
          scan1.setCacheBlocks(false);
  
          hBaseConf.set(TableInputFormat.INPUT_TABLE, "FFTransFlat");
          hBaseConf.set(TableInputFormat.SCAN, TableMapReduceUtil.convertScanToString(scan1));
  
          JavaPairRDD&lt;ImmutableBytesWritable, Result&gt; source = sc
                  .newAPIHadoopRDD(hBaseConf, TableInputFormat.class,
                          ImmutableBytesWritable.class, Result.class);
  
          System.out.println(source.count());
  
          // Test 3: bulkPut()  
  
           List&lt;String&gt; list = new ArrayList&lt;String&gt;(5);
              list.add("1,CI,a,1");   
              list.add("2,CI,a,2");   
              list.add("3,CI,a,3");   
  
              JavaRDD&lt;String&gt; rdd = sc.parallelize(list);
  
              byte tableName[] = "FFTransFlat".getBytes();
              hbaseContext.bulkPut(rdd,
                      TableName.valueOf(tableName),
                      new PutFunction());
  
              System.out.println(source.count());
  
  
          connection.close();
      }
  
  }
  </code></pre>
  
  <p>I see from the DataProc website that Spark 2.2.0 and 1.6.2 are supported. I had issues with 2.2.0, so I'm using 1.6.2.</p>
  
  <p>Can I get some advice on the following:
  What is the right combination of dependencies and versions to use (specifically to work with Cloud Bigtable, not HBase cluster)</p>
  
  <p>Is the recommended to achieve parallelization by using <code>newAPIHadoopRDD</code> or something like <code>bulkRead()/bulkDelete()</code>/etc. ? Or is there another preferred and performant way to do MPP with DataProc/Bigtable?</p>
  
  <p>Apologies for the lengthy post -- this is our first DataProc attempt.</p>
  
  <p>*** UPDATE:</p>
  
  <p>I managed to get something working after updating the Bigtable dependency to <code>bigtable-hbase-2.x-hadoop</code> and the HBase versions to <code>2.0.0-alpha2</code>. At least the bulkPut seems to be working at this stage. Now will work on cleaning out the unneeded stuff from dependencies.</p>