Spark与mongodb的结合

旧城等待， 2022-06-02 05:56 172阅读 0赞

## Mongodb ##

> Mongodb的安装  
> 实验环境：ubuntu16.04

安装：suso apt-get install mongodb
    查看版本：mongo -version
    启动和关闭mongodb
    service mongodb start
    service mongodb stop
    shell命令使用mongodb:   mongo

## Spark-MongoDB ##

[MongoDB-Spark连机器官网][MongoDB-Spark]

import com.mongodb.spark.MongoSpark;
    import com.mongodb.spark.config.WriteConfig;
    
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.sql.SparkSession;
    import org.bson.Document;
    import static java.util.Arrays.asList;
    import java.util.HashMap;
    import java.util.Map;
    
    public final class WriteToMongoDBWriteConfig {
    
      public static void main(final String[] args) throws InterruptedException {
    
        SparkSession spark = SparkSession.builder()
          .master("local")
          .appName("MongoSparkConnectorIntro")
          .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/Spark.myCollection")
          .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/Spark.myCollection")
          .getOrCreate();
    
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
    
        // Create a custom WriteConfig
        Map<String, String> writeOverrides = new HashMap<String, String>();
        writeOverrides.put("collection", "spark");
        writeOverrides.put("writeConcern.w", "majority");
        WriteConfig writeConfig = WriteConfig.create(jsc).withOptions(writeOverrides);
    
        JavaRDD<Document> sparkDocuments = jsc.parallelize(asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).map
             (new Function<Integer, Document>() {
           public Document call(final Integer i) throws Exception {
                   return Document.parse("{spark: " + i + "}");
                 }
          });
              MongoSpark.save(sparkDocuments, writeConfig);
    
        jsc.close();
    
      }
    
    }

.config("spark.mongodb.input.uri", "mongodb://127.0.0.1/Spark.myCollection")
    .config("spark.mongodb.output.uri", "mongodb://127.0.0.1/Spark.myCollection")
          .getOrCreate();
    此处的Spark是MongoDB的数据库名，myCollection类似MySQL的表名

[MongoDB-Spark]: https://docs.mongodb.com/spark-connector/v2.0/java-api/