read_jdbc_parquet_write_mongo.scala

def getSparkSessionMongoDbConfig(parms: Map[String, String]): SparkSession = {
    val spark = SparkSession
      .builder
      .appName(parms("JOB_NAME"))
      .master("local[*]")
      .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/retaildb.orders?authSource=admin")
      .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/retaildb.orders?authSource=admin")
      .getOrCreate()

    val isS3Enable = parms("S3_OPERATION_ENABLE").toBoolean;
    if (isS3Enable) {
      spark.sparkContext.hadoopConfiguration.set("fs.s3a.access.key", parms("S3_ACCESS_KEY"))
      spark.sparkContext.hadoopConfiguration.set("fs.s3a.secret.key", parms("S3_SECRET_KEY"))
      spark.sparkContext.hadoopConfiguration.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
      spark.sparkContext.hadoopConfiguration.set("spark.speculation", "false")
    }
    return spark;
  }

val fileName = "config/app_mongodb.properties";
    val params: Map[String, String] = com.mts.analytics.utils.Utils.getConfig(fileName);
    logger.info("File Name :" + params)
    val spark = com.mts.analytics.utils.SparkUtils.getSparkSessionMongoDbConfig(params);
    println("spark :" + spark)

    // Read From File Parquet
    val readFromParquetDF = spark.read.format("parquet").load("C:/data/mts_data/");

    val writeMongoDF = spark.read.format("jdbc")
      .option("url", params("JDBC_URL"))
      .option("dbtable", "products")
      .option("user", params("JDBC_USERNAME"))
      .option("password", params("JDBC_PASSWORD")).load()

    writeMongoDF.printSchema()
    writeMongoDF.show(10, false)

    //Write into MongoDB retaildb.products
    writeMongoDF.write.format("com.mongodb.spark.sql.DefaultSource").mode("append").option("database", "retaildb").option("collection", "products").save();
    print("Write Into MongoDB DataFrame completed ")


    val readMongoDF = spark.read.format("com.mongodb.spark.sql.DefaultSource").option("uri", "mongodb://127.0.0.1/retaildb.products?authSource=admin").load()
    readMongoDF.printSchema()
    readMongoDF.show(10, false)