happy_boy 发表于 2017-12-16 10:47:48

在mongodb中存值,如果存在则修改,不存在插入

import com.mongodb.hadoop.MongoOutputFormat  
import org.apache.hadoop.conf.Configuration
  
import org.apache.spark.rdd.RDD
  
import org.apache.spark.{SparkConf, SparkContext}
  
import org.bson.BasicBSONObject
  
import com.mongodb.{BasicDBObject}
  
import com.mongodb.hadoop.io.MongoUpdateWritable
  
/**
  * Created by Administrator on 2016/8/23 0023.
  */
  
object TestMongoDB {
  def main(args: Array): Unit = {
  System.setProperty("hadoop.home.dir", "E:/hadoop-2.4.1/")
  val conf = new SparkConf().setAppName("macType").setMaster("local")
  val sc = new SparkContext(conf)
  var data = sc.textFile("F:/aa.txt")
  insertResultToMongoDB(data)
  }
  def insertResultToMongoDB(rdd1:RDD): Unit ={
  val config = new Configuration()
  config.set("mongo.output.uri", "mongodb://localhost:27017/admin.zz") //输出表名
  val rdd = rdd1.map((s: String) =>{
  var bson = new BasicBSONObject()
  var time = System.currentTimeMillis()
  bson.put("_id",s)
  bson.put("xxx", time.toString)
  (s,new MongoUpdateWritable(
  new BasicDBObject("_id", s),// Query
  new BasicDBObject("$set", bson),// Update operation
  true,// Upsert
  false,   // Update multiple documents
  false
  ))
  })
  for(e <- rdd){
  println(e._1+"" + e._2)
  }
  println("-------------------")
  rdd.saveAsNewAPIHadoopFile(
  "file:///this-is-completely-unused",
  classOf,
  classOf,
  classOf],
  config)
  }
  
}
  
页: [1]
查看完整版本: 在mongodb中存值,如果存在则修改,不存在插入