zoukankan      html  css  js  c++  java
  • mongo-spark-读取不同的库数据和写入不同的库中

    mongo-spark-读取不同的库数据和写入不同的库中

     package com.example.app
    
     import com.mongodb.spark.config.{ReadConfig, WriteConfig}
     import com.mongodb.spark.sql._
    
    object App {
    
    
     def main(args: Array[String]): Unit = {
    
        val MongoUri1 = args(0).toString
        val MongoUri2 = args(1).toString
        val SparkMasterUri= args(2).toString
    
         def makeMongoURI(uri:String,database:String,collection:String) = (s"${uri}/${database}.${collection}")
    
       val mongoURI1 = s"mongodb://${MongoUri1}:27017"
       val mongoURI2 = s"mongodb://${MongoUri2}:27017"
    
       val CONFdb1 = makeMongoURI(s"${mongoURI1}","MyColletion1,"df")
       val CONFdb2 = makeMongoURI(s"${mongoURI2}","MyColletion2,"df")
    
       val WRITEdb1: WriteConfig =  WriteConfig(scala.collection.immutable.Map("uri"->CONFdb1))
       val READdb1: ReadConfig = ReadConfig(Map("uri" -> CONFdb1))
    
       val WRITEdb2: WriteConfig =  WriteConfig(scala.collection.immutable.Map("uri"->CONFdb2))
       val READdb2: ReadConfig = ReadConfig(Map("uri" -> CONFdb2))
    
       val spark = SparkSession
      .builder
      .appName("AppMongo")
      .config("spark.worker.cleanup.enabled", "true")
      .config("spark.scheduler.mode", "FAIR")
      .getOrCreate()
    
       val df1 = spark.read.mongo(READdb1)
       val df2 = spark.read.mongo(READdb2)
       df1.write.mode("overwrite").mongo(WRITEdb1)
       df2.write.mode("overwrite").mongo(WRITEdb2)
     }
    }
    
  • 相关阅读:
    MySQL行级锁、表级锁、页级锁详细介绍
    Spring REST是什么?(转)
    Spring REST(转)
    联系人项目
    三级联动(有刷新)
    Java中点击按钮返回上一页
    Java中隐藏显示效果
    理解RESTful架构(转)
    什么是REST?以及RESTful的实现(转)
    JSTL 核心标签库 使用(转)
  • 原文地址:https://www.cnblogs.com/wenbronk/p/9887466.html
Copyright © 2011-2022 走看看