title: Spark 2.2.0 - How to write/read DataFrame to DynamoDB
body: <p>I want my Spark application to read a table from DynamoDB, do stuff, then write the result in DynamoDB. </p>
  
  <h1>Read the table into a DataFrame</h1>
  
  <p>Right now, I can read the table from DynamoDB into Spark as a <code>hadoopRDD</code> and convert it to a DataFrame. However, I had to use a regular expression to extract the value from <code>AttributeValue</code>. Is there a better/more elegant way? Couldn't find anything in the AWS API.</p>
  
  <pre><code>package main.scala.util
  
  import org.apache.spark.sql.SparkSession
  import org.apache.spark.SparkContext
  import org.apache.spark.sql.SQLContext
  import org.apache.spark.sql.functions._
  import org.apache.spark.sql.types._
  import org.apache.spark.rdd.RDD
  import scala.util.matching.Regex
  import java.util.HashMap
  
  import com.amazonaws.services.dynamodbv2.model.AttributeValue
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.dynamodb.DynamoDBItemWritable
  /* Importing DynamoDBInputFormat and DynamoDBOutputFormat */
  import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
  import org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat
  import org.apache.hadoop.mapred.JobConf
  import org.apache.hadoop.io.LongWritable
  
  object Tester {
  
    // {S: 298905396168806365,} 
    def extractValue : (String =&gt; String) = (aws:String) =&gt; {
      val pat_value = "\\s(.*),".r
  
      val matcher = pat_value.findFirstMatchIn(aws)
                  matcher match {
                  case Some(number) =&gt; number.group(1).toString
                  case None =&gt; ""
          }
    }
  
  
     def main(args: Array[String]) {
      val spark = SparkSession.builder().getOrCreate()
      val sparkContext = spark.sparkContext
  
        import spark.implicits._
  
        // UDF to extract Value from AttributeValue 
        val col_extractValue = udf(extractValue)
  
    // Configure connection to DynamoDB
    var jobConf_add = new JobConf(sparkContext.hadoopConfiguration)
        jobConf_add.set("dynamodb.input.tableName", "MyTable")
        jobConf_add.set("dynamodb.output.tableName", "MyTable")
        jobConf_add.set("mapred.output.format.class", "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
        jobConf_add.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")
  
  
        // org.apache.spark.rdd.RDD[(org.apache.hadoop.io.Text, org.apache.hadoop.dynamodb.DynamoDBItemWritable)]
        var hadooprdd_add = sparkContext.hadoopRDD(jobConf_add, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
  
        // Convert HadoopRDD to RDD
        val rdd_add: RDD[(String, String)] = hadooprdd_add.map {
        case (text, dbwritable) =&gt; (dbwritable.getItem().get("PIN").toString(), dbwritable.getItem().get("Address").toString())
        }
  
        // Convert RDD to DataFrame and extract Values from AttributeValue
        val df_add = rdd_add.toDF()
                    .withColumn("PIN", col_extractValue($"_1"))
                    .withColumn("Address", col_extractValue($"_2"))
                    .select("PIN","Address")
     }
  }
  </code></pre>
  
  <h1>Write the DataFrame to DynamoDB</h1>
  
  <p>Many answers in stackoverflow and elsewhere only point to the <a href="https://aws.amazon.com/blogs/big-data/analyze-your-data-on-amazon-dynamodb-with-apache-spark/" rel="noreferrer">blog post</a> and the <a href="https://github.com/awslabs/emr-dynamodb-connector" rel="noreferrer">emr-dynamodb-hadoop github</a>. None of those resources actually demonstrate how to write to DynamoDB. </p>
  
  <p><a href="https://stackoverflow.com/questions/43248940/how-to-convert-dataframe-in-spark-to-hadooprdd">I tried converting</a> my <code>DataFrame</code> to <code>RDD[Row]</code> unsuccessfully.</p>
  
  <pre><code>df_add.rdd.saveAsHadoopDataset(jobConf_add)
  </code></pre>
  
  <p>What are the steps to write this DataFrame to DynamoDB? (Bonus Points if you tell me how to control <code>overwrite</code> vs <code>putItem</code> ;)</p>
  
  <p>Note: <code>df_add</code> has the same schema as <code>MyTable</code> in DynamoDB.</p>
  
  <p><strong>EDIT</strong>: I am following the recommendation from <a href="https://stackoverflow.com/questions/35733968/simple-rdd-write-to-dynamodb-in-spark">this answer</a> which points to this post on <a href="https://aws.amazon.com/blogs/big-data/using-spark-sql-for-etl/" rel="noreferrer">Using Spark SQL for ETL</a>:</p>
  
  <pre><code>// Format table to DynamoDB format
    val output_rdd =  df_add.as[(String,String)].rdd.map(a =&gt; {
      var ddbMap = new HashMap[String, AttributeValue]()
  
      // Field PIN
      var PINValue = new AttributeValue() // New AttributeValue
      PINValue.setS(a._1)                 // Set value of Attribute as String. First element of tuple
      ddbMap.put("PIN", PINValue)         // Add to HashMap
  
      // Field Address
      var AddValue = new AttributeValue() // New AttributeValue
      AddValue.setS(a._2)                 // Set value of Attribute as String
      ddbMap.put("Address", AddValue)     // Add to HashMap
  
      var item = new DynamoDBItemWritable()
      item.setItem(ddbMap)
  
      (new Text(""), item)
    })             
  
    output_rdd.saveAsHadoopDataset(jobConf_add) 
  </code></pre>
  
  <p>However, now I am getting <code>java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.hadoop.io.Text</code> despite following the documentation ... Do you have any suggestion ? </p>
  
  <p><strong>EDIT 2</strong>: Reading more carefully this post on <a href="https://aws.amazon.com/blogs/big-data/using-spark-sql-for-etl/" rel="noreferrer">Using Spark SQL for ETL</a>:</p>
  
  <blockquote>
    <p>After you have the DataFrame, perform a transformation to have an RDD that matches the types that the DynamoDB custom output format knows how to write.  The custom output format expects a tuple containing the Text and <code>DynamoDBItemWritable</code> types.</p>
  </blockquote>
  
  <p>Taking this into account, the code below is exactly what theAWS blog post suggest, except I cast <code>output_df</code> as an rdd otherwise <code>saveAsHadoopDataset</code> doesn't work. And now, I am getting <code>Exception in thread "main" scala.reflect.internal.Symbols$CyclicReference: illegal cyclic reference involving object InterfaceAudience</code>. I am at the end of my rope!</p>
  
  <pre><code>      // Format table to DynamoDB format
    val output_df =  df_add.map(a =&gt; {
      var ddbMap = new HashMap[String, AttributeValue]()
  
      // Field PIN
      var PINValue = new AttributeValue() // New AttributeValue
      PINValue.setS(a.get(0).toString())                 // Set value of Attribute as String
      ddbMap.put("PIN", PINValue)         // Add to HashMap
  
      // Field Address
      var AddValue = new AttributeValue() // New AttributeValue
      AddValue.setS(a.get(1).toString())                 // Set value of Attribute as String
      ddbMap.put("Address", AddValue)     // Add to HashMap
  
      var item = new DynamoDBItemWritable()
      item.setItem(ddbMap)
  
      (new Text(""), item)
    })             
  
    output_df.rdd.saveAsHadoopDataset(jobConf_add)   
  </code></pre>