Find Data in DynamoDB in Spark

Find Data in DynamoDB in Spark

I get a small task recently, in our DynamoDB, there is one column in one table, at first we put bad data in table saying extension as Number, later we know it should be String.
But that end up we have some Number and some String in the same column. We need to find out all the Number ones and update them to String.
Here is my steps.
First of all I do a scan
> aws dynamodb scan --table-name sillycat_device-stage-devicePairingInfo --query "Items

.[extension.N]" --output text > ./extensionnumber.csv

In the file extension number.csv, I will have data similar to
extension
None
None
None
None
None
None
243074
I need filter out all the None ones, then I get the Number ones.
Then I dump the whole table
> export-dynamodb -t sillycat_device-stage-devicePairingInfo -f csv -o devicepairinginfodb.csv
Put these 2 tables in Spark, do a join and select, I found out all the ones with Number.
In Spark, we can directly read DynamoDB if they do not have this kind of Number/String conflicts.
import com.github.traviscrawford.spark.dynamodb._
%spark.dep
z.load("mysql:mysql-connector-java:5.1.47")
z.load("com.github.traviscrawford:spark-dynamodb:0.0.13”)
val accountDF = sqlContext.read.dynamodb("us-west-1", “sillycat_device-stage-devicePairingInfo")
accountDF.printSchema()
accountDF.registerTempTable("devicepairing")
Or load the CSV file
val devicePairingDF = sqlContext.read.format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("hdfs://localhost:9000/convertdevicepairingextension/devicepairinginfodb.csv") 
devicePairingDF.printSchema()
devicePairingDF.createOrReplaceTempView("devicepairing2")
Load the second Number file
val extensionRawDF = sqlContext.read.format("csv")
  .option("header", "true")
  .option("inferSchema", "true")
  .load("hdfs://localhost:9000/convertdevicepairingextension/extensionnumber2.csv") 
val extensionRaw1DF = extensionRawDF.toDF(extensionRawDF.columns map(_.toLowerCase): _*)
val extensionDF = extensionRaw1DF.columns.foldLeft(extensionRaw1DF)((curr, n) => curr.withColumnRenamed(n, n.replaceAll("\\s", "_")))
extensionDF.printSchema()
extensionDF.createOrReplaceTempView("extension”)
Join the 2 tables
%sql
select t1.* from devicepairing2 t1, extension t2 where t1.extension = t2.extension
Output the JSON file
val extensionUpdateDF = sqlContext.sql("""
select t1.* from devicepairing2 t1, extension t2 where t1.extension = t2.extension
""")
extensionUpdateDF.show(2)
extensionUpdateDF.repartition(1).write.json("hdfs://localhost:9000/convertdevicepairingextension/extension_to_update")

References: