スパーク：インポートデータフレームのMongoDBする（スカラ）

Question

スパークで、次のデータフレームを考える：

Name,LicenseID_1,TypeCode_1,State_1,LicenseID_2,TypeCode_2,State_2,LicenseID_3,TypeCode_3,State_3  
"John","123ABC",1,"WA","456DEF",2,"FL","789GHI",3,"CA" 
"Jane","ABC123",5,"AZ","DEF456",7,"CO","GHI789",8,"GA" 

がどのように私は次のように文書のコレクションとしてのMongoDBにこれを書くためにスパークでのScalaを使用することができます。

{ "Name" : "John", 
    "Licenses" : 
    { 
    [ 
     {"LicenseID":"123ABC","TypeCode":"1","State":"WA" }, 
     {"LicenseID":"456DEF","TypeCode":"2","State":"FL" }, 
     {"LicenseID":"789GHI","TypeCode":"3","State":"CA" } 
    ] 
    } 
}, 

{ "Name" : "Jane", 
    "Licenses" : 
    { 
    [ 
     {"LicenseID":"ABC123","TypeCode":"5","State":"AZ" }, 
     {"LicenseID":"DEF456","TypeCode":"7","State":"CO" }, 
     {"LicenseID":"GHI789","TypeCode":"8","State":"GA" } 
    ] 
    } 
} 

私はこれを実行しようとしましたが、次のコードのブロックを得た：

val customSchema = StructType(Array(StructField("Name", StringType, true), StructField("LicenseID_1", StringType, true), StructField("TypeCode_1", StringType, true), StructField("State_1", StringType, true), StructField("LicenseID_2", StringType, true), StructField("TypeCode_2", StringType, true), StructField("State_2", StringType, true), StructField("LicenseID_3", StringType, true), StructField("TypeCode_3", StringType, true), StructField("State_3", StringType, true))) 
val license = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").schema(customSchema).load("D:\\test\\test.csv") 
case class License(LicenseID:String, TypeCode:String, State:String) 
case class Data(Name:String, Licenses: Array[License]) 
val transformedData = license.map(data => Data(data(0),Array(License(data(1),data(2),data(3)),License(data(4),data(5),data(6)),License(data(7),data(8),data(9))))) 

<console>:46: error: type mismatch; 
found : Any 
required: String 
     val transformedData = license.map(data => Data(data(0),Array(License(data(1),data(2),data(3)),License(data(4),data(5),data(6)),License(data(7),data(8),data(9))))) 
... 

Answer 1

ないあなたが何であるかを確認してくださいデータ火花やマンゴー

• https://docs.mongodb.com/spark-connector/current/

• https://docs.mongodb.com/spark-connector/current/scala-api/

  輸入org.apache.spark.sql.SparkSession 輸入com.mongodb.spark.sql._を保存する方法の例を追加し、Rの質問、

  val sc：SparkContext //既存のSparkContext。 ヴァルsparkSession = SparkSession.builder（）。getOrCreate（）

  //モンゴスパークヘルパー ヴァルDF = MongoSpark.load（sparkSession）//は読む

  SparkConf

を使用します

sparkSession.loadFromMongoDB() // Uses the SparkConf for configuration sparkSession.loadFromMongoDB(ReadConfig(Map("uri" -> "mongodb://example.com/database.collection"))) // Uses the ReadConfig sparkSession.read.mongo() sparkSession.read.format("com.mongodb.spark.sql").load() // Set custom options: sparkSession.read.mongo(customReadConfig) sparkSession.read.format("com.mongodb.spark.sql").options. (customReadConfig.asOptions).load() 

コネクタは、データをMongoDBに保持する機能を提供します。

MongoSpark.save(centenarians.write.option("collection", "hundredClub")) 
    MongoSpark.load[Character](sparkSession, ReadConfig(Map("collection" -> 

"データ"）、いくつかの（ReadConfig（sparkSession））））。ショー（）

代替が保存するデータ

dataFrameWriter.write.mongo() 
dataFrameWriter.write.format("com.mongodb.spark.sql").save() 

Answer 2

.toStringは、問題を修正し、私ができたの追加私が望む形式をmongodbに保存しました。

val transformedData = license.map（data => Data（0）.toString、Array（ライセンスデータ（1）.toString、data（2）.toString、data（3）.toString）、License （データ（7）.toString、データ（8）.toString、データ（9）.toString））））））））））））