title: How to read data from dynamo db table into dataframe?
body: <p>Below is the code where I am trying to read data from dynamo db and load it into a dataframe.</p>
  
  <p>Is it possible to do the same using scanamo? </p>
  
  <pre><code>import org.apache.hadoop.io.Text;
  import org.apache.hadoop.dynamodb.DynamoDBItemWritable
  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
  
  var jobConf = new JobConf(sc.hadoopConfiguration)
  jobConf.set("dynamodb.servicename", "dynamodb")
  jobConf.set("dynamodb.input.tableName", "GenreRatingCounts")   // Pointing to DynamoDB table
  jobConf.set("dynamodb.endpoint", "dynamodb.us-east-2.amazonaws.com")
  jobConf.set("dynamodb.regionid", "us-east-2")
  jobConf.set("dynamodb.throughput.read", "1")
  jobConf.set("dynamodb.throughput.read.percent", "1")
  jobConf.set("dynamodb.version", "2011-12-05")
  jobConf.set("dynamodb.awsAccessKeyId", "XXXXX")
  jobConf.set("dynamodb.awsSecretAccessKey", "XXXXXXX")
  jobConf.set("mapred.output.format.class", "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
  jobConf.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")
  
  var orders = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
  
  orders.map(t =&gt; t._2.getItem()).collect.foreach(println)    
  val simple2: RDD[(String)] = orders.map { case (text, dbwritable) =&gt; (dbwritable.toString)}
  spark.read.json(simple2).registerTempTable("gooddata")
  </code></pre>
  
  <p>The output is of type: <code>org.apache.spark.sql.DataFrame = [count: struct&lt;n: string&gt;, genre: struct&lt;s: string&gt; ... 1 more field]</code></p>
  
  <pre><code>+------+---------+------+
  | count|    genre|rating|
  +------+---------+------+
  |[4450]| [Action]|   [4]|
  |[5548]|[Romance]| [3.5]|
  +------+---------+------+
  </code></pre>
  
  <p>How can I convert this dataframe column types to String instead of Struct? </p>
  
  <p>EDIT-1</p>
  
  <p>Now I am able to create dataframe using below code and able to read data from dynamodb table if it doesn't contain null. </p>
  
  <pre><code>var orders = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])
  
  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; ""
          }
    }
  val col_extractValue = udf(extractValue)
  
  val rdd_add = orders.map {
        case (text, dbwritable) =&gt; (dbwritable.getItem().get("genre").toString(), dbwritable.getItem().get("rating").toString(),dbwritable.getItem().get("ratingCount").toString())
  
  val df_add = rdd_add.toDF()
                    .withColumn("genre", col_extractValue($"_1"))
                    .withColumn("rating", col_extractValue($"_2"))
                    .withColumn("ratingCount", col_extractValue($"_3"))
                    .select("genre","rating","ratingCount")
  
  df_add.show
  </code></pre>
  
  <p>But I am getting below error if there is a record with no data in one of the column(null or blank). </p>
  
  <pre><code> ERROR Executor: Exception in task 0.0 in stage 10.0 (TID 14)
  java.lang.NullPointerException
          at $line117.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(&lt;console&gt;:67)
          at $line117.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(&lt;console&gt;:66)
          at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
          at scala.collection.Iterator$class.foreach(Iterator.scala:891)
          at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
          at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
          at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
          at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
          at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
          at scala.collection.AbstractIterator.to(Iterator.scala:1334)
          at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
          at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
          at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
          at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
          at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
          at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
          at org.apache.spark.scheduler.Task.run(Task.scala:123)
          at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
          at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  19/12/20 07:48:21 WARN TaskSetManager: Lost task 0.0 in stage 10.0 (TID 14, localhost, executor driver): java.lang.NullPointerException
          at $line117.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(&lt;console&gt;:67)
          at $line117.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(&lt;console&gt;:66)
          at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
          at scala.collection.Iterator$class.foreach(Iterator.scala:891)
          at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
          at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
          at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
          at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
          at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
          at scala.collection.AbstractIterator.to(Iterator.scala:1334)
          at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
          at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1334)
          at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
          at scala.collection.AbstractIterator.toArray(Iterator.scala:1334)
          at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
          at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:945)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
          at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
          at org.apache.spark.scheduler.Task.run(Task.scala:123)
          at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
          at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  
  <p>How to handle null/blank while reading from Dynamodb to a dataframe in Spark/Scala?</p>