title: How to use Spark's .newAPIHadoopRDD() from Java
body: <p>I am trying to port an example written in Scala (from the Apache Spark project) into Java, and running into some issues.</p>
  
  <p>The code </p>
  
  <pre><code>val casRdd = sc.newAPIHadoopRDD(job.getConfiguration(),
    classOf[CqlPagingInputFormat],
    classOf[java.util.Map[String,ByteBuffer]],
    classOf[java.util.Map[String,ByteBuffer]])
  </code></pre>
  
  <p>from the original Scala example builds and runs just fine, but</p>
  
  <pre><code>JavaPairRDD rdd = jsc.newAPIHadoopRDD(job.getConfiguration(),
    CqlPagingInputFormat.class,
    java.util.Map&lt;String, ByteBuffer&gt;.class,
    java.util.Map&lt;String, ByteBuffer&gt;.class);
  </code></pre>
  
  <p>is not allowed in Java (<code>Cannot select from parameterized type</code>).</p>
  
  <p>Changing</p>
  
  <pre><code>java.util.Map&lt;String, ByteBuffer&gt;.class
  </code></pre>
  
  <p>into</p>
  
  <pre><code>Class.forName("java.util.Map&lt;String, ByteBuffer&gt;")
  </code></pre>
  
  <p>yields a new error:</p>
  
  <pre><code>Error:(42, 30) java: method newAPIHadoopRDD in class org.apache.spark.api.java.JavaSparkContext cannot be applied to given types;
  required: org.apache.hadoop.conf.Configuration,java.lang.Class&lt;F&gt;,java.lang.Class&lt;K&gt;,java.lang.Class&lt;V&gt;
  found: org.apache.hadoop.conf.Configuration,java.lang.Class&lt;org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat&gt;,java.lang.Class&lt;capture#1 of ?&gt;,java.lang.Class&lt;capture#2 of ?&gt;
  reason: inferred type does not conform to declared bound(s)
  inferred: org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat
  bound(s): org.apache.hadoop.mapreduce.InputFormat&lt;capture#1 of ?,capture#2 of ?&gt;
  </code></pre>
  
  <p>Changing it into simply <code>java.util.Map.class</code> yields a similar error:</p>
  
  <pre><code>Error:(44, 30) java: method newAPIHadoopRDD in class org.apache.spark.api.java.JavaSparkContext cannot be applied to given types;
  required: org.apache.hadoop.conf.Configuration,java.lang.Class&lt;F&gt;,java.lang.Class&lt;K&gt;,java.lang.Class&lt;V&gt;
  found: org.apache.hadoop.conf.Configuration,java.lang.Class&lt;org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat&gt;,java.lang.Class&lt;java.util.Map&gt;,java.lang.Class&lt;java.util.Map&gt;
  reason: inferred type does not conform to declared bound(s)
  inferred: org.apache.cassandra.hadoop.cql3.CqlPagingInputFormat
  bound(s): org.apache.hadoop.mapreduce.InputFormat&lt;java.util.Map,java.util.Map&gt;
  </code></pre>
  
  <p>So what is the correct translation? Worth noting that the <code>newAPIHadoopRDD()</code> function is a different implementation for Scala and for Java. Documentation for the methods can be found <a href="http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext" rel="nofollow">here</a> for Scala and here: <a href="http://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaSparkContext.html#newAPIHadoopRDD%28org.apache.hadoop.conf.Configuration" rel="nofollow">http://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaSparkContext.html#newAPIHadoopRDD(org.apache.hadoop.conf.Configuration</a>, java.lang.Class, java.lang.Class, java.lang.Class) for Java.</p>
  
  <p>The declaration of <code>CqlPagingInputFormat</code> looks like this</p>
  
  <pre><code>public class CqlPagingInputFormat extends org.apache.cassandra.hadoop.AbstractColumnFamilyInputFormat&lt;java.util.Map&lt;java.lang.String,java.nio.ByteBuffer&gt;,java.util.Map&lt;java.lang.String,java.nio.ByteBuffer&gt;&gt; {
  </code></pre>