title: Spark Streaming - Java - Insert JSON from Kafka into Cassandra
body: <p>I'm writing a simple data pipeline in Spark Streaming, using Java, to pull JSON data from Kafka, parse the JSON into a custom class (<code>Transaction</code>), then insert that data into a Cassandra table but I am unable to get the <code>mapToRow()</code> function to work.</p>
  
  <p>I've seen tons of examples that say all you have to do is something along the lines of this:</p>
  
  <pre><code>JavaPairInputDStream&lt;String, String&gt; stream = KafkaUtils.createDirectStream(
          streamingContext,
          String.class, 
          String.class, 
          StringDecoder.class, 
          StringDecoder.class,
          kafkaParams,
          topicsSet
  );
  
  JavaDStream&lt;String&gt; lines = stream.map(
      new Function&lt;Tuple2&lt;String,String&gt;, String&gt;(){
          @Override
          public String call(Tuple2&lt;String,String&gt; tuple2) {
              return tuple2._2();
          }
      }
  );
  
  javaFunctions(lines).writerBuilder("myKeyspace", "myTableName", mapToRow(Transaction.class)).saveToCassandra();
  </code></pre>
  
  <p>However, when I do this I get the error:</p>
  
  <pre><code>The method mapToRow(Class&lt;Transaction&gt;) is undefined for the type SaveTransactions
  </code></pre>
  
  <p>I think all I am lacking is some sort of decoration on my class but I have not been successful in figuring out which one.  I've tried going bare bones, essentially making the class a property bag:</p>
  
  <pre><code>public class Transaction implements java.io.Serializable{
  
      public int TransactionId;
      ...
  
      public Transaction(){}
  }
  </code></pre>
  
  <p>I've tried all of the DataStax mapping annotations:</p>
  
  <pre><code>@Table(keyspace = "myKeyspace", name = "myTableName",
         readConsistency = "QUORUM",
         writeConsistency = "QUORUM",
         caseSensitiveKeyspace = false,
         caseSensitiveTable = false)
  public class Transaction implements java.io.Serializable{
  
      @PartitionKey(0)
      @Column(name="transaction_id")
      public int TransactionId;
      ...
  
      public Transaction(){}
  }
  </code></pre>
  
  <p>I also tried establishing public get/set methods for each property and setting the properties to private:</p>
  
  <pre><code>public class Transaction implements java.io.Serializable{
  
      private int transactionId;
      ...
  
      public Transaction(){}
  
      public int getTransactionId() {
          return transactionId;
      }
  
      public void setTransactionId(int transactionId) {
          this.transactionId = transactionId;
      }
  }
  </code></pre>
  
  <p>I have been able to parse the <code>DStream</code> into an <code>RDD</code> of <code>Transactions</code> using the class below:</p>
  
  <pre><code>public class Transaction implements java.io.Serializable{
  
      ...
  
      public static class ParseJSON implements FlatMapFunction&lt;Iterator&lt;String&gt;, Transaction&gt; {
          public Iterable&lt;Transaction&gt; call(Iterator&lt;String&gt; lines) throws Exception {
              ArrayList&lt;Transaction&gt; transactions = new ArrayList&lt;Transaction&gt;();
                  ObjectMapper mapper = new ObjectMapper();
                  while (lines.hasNext()) {
                      String line = lines.next();
                      try {
                          transactions.add(mapper.readValue(line, Transaction.class));
                      } catch (Exception e) {
                          System.out.println("Skipped:" + e);
                      }
                  }
  
                  return transactions;
          }
      }
  }
  </code></pre>
  
  <p>In conjunction with the following code, acting on the <code>lines</code> object from above:</p>
  
  <pre><code>JavaDStream&lt;Transaction&gt; events = lines.mapPartitions(new Transaction.ParseJSON());
  </code></pre>
  
  <p>However, once I have it in this from it still doesn't work with the writeBuilder().saveToCassandra() chain.</p>
  
  <p>Any help here is greatly appreciated.</p>