我希望我的 Spark 应用程序从 DynamoDB 读取表，执行操作，然后将结果写入 DynamoDB。
将表读入 DataFrame
现在，我可以将表从 DynamoDB 读入 SparkhadoopRDD并将其转换为 DataFrame。但是，我必须使用正则表达式来提取值AttributeValue。有更好/更优雅的方式吗？在 AWS API 中找不到任何内容。
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 => String) = (aws:String) => {
val pat_value = "\\s(.*),".r
val matcher = pat_value.findFirstMatchIn(aws)
matcher match {
case Some(number) => number.group(1).toString
case None => ""
}
}
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) => (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")
}
}
将 DataFrame 写入 DynamoDB
stackoverflow 和其他地方的许多答案仅指向博客文章 https://aws.amazon.com/blogs/big-data/analyze-your-data-on-amazon-dynamodb-with-apache-spark/和emr-dynamodb-hadoop github https://github.com/awslabs/emr-dynamodb-connector。这些资源都没有实际演示如何写入 DynamoDB。
我尝试转换 https://stackoverflow.com/questions/43248940/how-to-convert-dataframe-in-spark-to-hadooprdd my DataFrame to RDD[Row]不成功。
df_add.rdd.saveAsHadoopDataset(jobConf_add)
将此 DataFrame 写入 DynamoDB 的步骤是什么？ （如果你告诉我如何控制的话，奖励积分overwrite vs putItem ;)
Note: df_add具有相同的架构MyTable在 DynamoDB 中。
EDIT: 我正在遵循以下建议这个答案 https://stackoverflow.com/questions/35733968/simple-rdd-write-to-dynamodb-in-spark指向这篇文章使用 Spark SQL 进行 ETL https://aws.amazon.com/blogs/big-data/using-spark-sql-for-etl/:
// Format table to DynamoDB format
val output_rdd = df_add.as[(String,String)].rdd.map(a => {
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)
然而，现在我得到了java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.hadoop.io.Text尽管遵循文档...您有什么建议吗？
EDIT 2: 仔细阅读这篇文章使用 Spark SQL 进行 ETL https://aws.amazon.com/blogs/big-data/using-spark-sql-for-etl/:
获得 DataFrame 后，执行转换以获得与 DynamoDB 自定义输出格式知道如何编写的类型相匹配的 RDD。自定义输出格式需要一个包含 Text 和DynamoDBItemWritable types.
考虑到这一点，下面的代码正是 AWS 博客文章所建议的，除了我强制转换output_df否则作为 rddsaveAsHadoopDataset不起作用。现在，我得到了Exception in thread "main" scala.reflect.internal.Symbols$CyclicReference: illegal cyclic reference involving object InterfaceAudience。我已是穷途末路了！
// Format table to DynamoDB format
val output_df = df_add.map(a => {
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)
