Merge remote-tracking branch 'upstream/branch-0.11'

Author: darroyo-stratio

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.11.1 (? 2016)
+
+* Max and Min splitVector bounds for not sharded collections (see doc)
+* Config parameter renamed idasobjectid -> idAsObjectId
+
 ## 0.11.0 (February 2016)
 
 * Compatibility with spark 1.6

doc/pom.xml

@@ -26,7 +26,7 @@
  <parent>
 	<groupId>com.stratio.datasource</groupId>
  <artifactId>spark-mongodb-parent</artifactId>
- <version>0.12.0-SNAPSHOT</version>
+ <version>0.11.2-RC1-SNAPSHOT</version>
  </parent>
  <properties>
  <jacoco.skip>true</jacoco.skip>

doc/src/site/sphinx/First_Steps.rst

@@ -91,12 +91,18 @@ Configuration parameters
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | schema_samplingRatio | 1.0 | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
-| writeConcern | mongodb.WriteConcern.ACKNOWLEDGED | No |
+| writeConcern | "safe"  | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | splitSize | 10 | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | splitKey | "fieldName" | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
+| splitKeyType | "dataTypeName" | No |
++-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
+| splitKeyMin | "minvalue" | No |
++-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
+| splitKeyMax | "maxvalue" | No |
++-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | credentials | "user,database,password;user,database,password" | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | updateFields | "fieldName,fieldName" | No |
@@ -117,7 +123,7 @@ Configuration parameters
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | threadsAllowedToBlockForConnectionMultiplier | "5" | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
-| idasobjectid | "false" | No |
+| idAsObjectId | "false" | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
 | connectionsTime | "180000" | No |
 +-----------------------------------------------+--------------------------------------------------------------------------------+-------------------------+
@@ -129,12 +135,95 @@ Configuration parameters
 
 **Note:** '_id' field is autogenerated in MongoDB, so by default, you can filter it as String. If you need a custom '_id', you have to set 'idasobjectid' property to "false" like in the above table.
 
+There are two ways to set up configuration:
+
+1. Using MongodbConfigBuilder, which should contains all the config with the right types.
+
+2. Using DataFrame API to read/write with a Map[String, String] or setting configuration from a SQL sentence (in String to String format) as in:
+
+ ::
+
+ CREATE TEMPORARY TABLE tableName USING com.stratio.datasource.mongodb
+ OPTIONS (host 'host:port', database 'highschool', collection 'students')"
+
+
+Credentials
+-----------
+
+To connect with credentials you should specify user, database and password.
+
+From MongodbConfigBuilder, you have to create a list of MongodbCredentials, here is an example:
+
+::
+
+ MongodbConfigBuilder(Map(Host -> List("localhost:27017"), Database -> "highschool", Collection ->"students",
+ List(com.stratio.datasource.mongodb.MongodbCredentials(user, database, password.toCharArray))
+ )).build
+
+
+In other case, (String format) you have to use the format set in the table above.
+
+ One credential:
+
+ ::
+
+ "user,database,password"
+
+
+
+ Two credentials:
+
+ ::
+
+ "user1,database1,password1;user2,database2,password2"
+
+
+
+SplitKey parameters
+-------------------
+
+An index is needed in the splitKey field.
+
+All splitKey parameters are optionals.
+
+ splitKey: Field to split for.
+
+ splitSize: Max size of each chunk in MB.
+
+If you want to use explicit boundaries to choose what data get from MongoDB, you will have to use these parameters:
+
+ - splitKeyType: Data type of splitKey field. Next MongoDB types are supported:
+ - "isoDate"
+ - "int"
+ - "long"
+ - "double"
+ - "string"
+
+ - splitKeyMin: Min value of the split in string format.
+
+ - splitKeyMax: Max value of the split in string format.
+
+ **Note:** Only data between boundaries would be available
+
+
 Examples
 ========
 
 Scala API
 ---------
 
+Launch the spark shell:
+::
+
+ $ bin/spark-shell --packages com.stratio.datasource:spark-mongodb_2.10:<VERSION>
+
+If you are using the spark shell, a SQLContext is already created and is available as a variable: 'sqlContext'.
+Alternatively, you could create a SQLContext instance in your spark application code:
+
+::
+
+ val sqlContext = new SQLContext(sc)
+
 To read a DataFrame from a Mongo collection, you can use the library by loading the implicits from `com.stratio.datasource.mongodb._`.
 
 To save a DataFrame in MongoDB you should use the saveToMongodb() function as follows:
@@ -172,8 +261,8 @@ In the example we can see how to use the fromMongoDB() function to read from Mon
  val readConfig = builder.build()
  val mongoRDD = sqlContext.fromMongoDB(readConfig)
  mongoRDD.registerTempTable("students")
- sqlContext.sql("SELECT name, age FROM students")
-
+ val dataFrame = sqlContext.sql("SELECT name, age FROM students")
+ dataFrame.show
 
 
 If you want to use a SSL connection, you need to add this 'import', and add 'SSLOptions' to the MongodbConfigBuilder:
@@ -191,9 +280,9 @@ Using StructType:
 
  import org.apache.spark.sql.types._
  val schemaMongo = StructType(StructField("name", StringType, true) :: StructField("age", IntegerType, true ) :: Nil)
- sqlContext.createExternalTable("mongoTable", "com.stratio.datasource.mongodb", schemaMongo, Map("host" -> "localhost:27017", "database" -> "highschool", "collection" -> "students"))
+ val df = sqlContext.read.schema(schemaMongo).format("com.stratio.datasource.mongodb").options(Map("host" -> "localhost:27017", "database" -> "highschool", "collection" -> "students")).load
+ df.registerTempTable("mongoTable")
  sqlContext.sql("SELECT * FROM mongoTable WHERE name = 'Torcuato'").show()
- sqlContext.sql("DROP TABLE mongoTable")
 
 
 Using DataFrameWriter:
@@ -242,9 +331,19 @@ Then:
 ::
 
  from pyspark.sql import SQLContext
- sqlContext.sql("CREATE TEMPORARY TABLE students_table USING com.stratio.datasource.mongodb OPTIONS (host 'host:port', database 'highschool', collection 'students')")
+ sqlContext.sql("CREATE TEMPORARY TABLE students_table USING com.stratio.datasource.mongodb OPTIONS (host 'localhost:27017', database 'highschool', collection 'students')")
  sqlContext.sql("SELECT * FROM students_table").collect()
 
+Using DataFrameReader and DataFrameWriter:
+::
+
+ df = sqlContext.read.format('com.stratio.datasource.mongodb').options(host='localhost:27017', database='highschool', collection='students').load()
+ df.select("name").collect()
+
+ df.select("name").write.format("com.stratio.datasource.mongodb").mode('overwrite').options(host='localhost:27017', database='highschool', collection='studentsview').save()
+ dfView = sqlContext.read.format('com.stratio.datasource.mongodb').options(host='localhost:27017', database='highschool', collection='studentsview').load()
+ dfView.show()
+
 Java API
 --------
 

pom.xml

@@ -21,7 +21,7 @@
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.stratio.datasource</groupId>
  <artifactId>spark-mongodb-parent</artifactId>
- <version>0.12.0-SNAPSHOT</version>
+ <version>0.11.2-RC1-SNAPSHOT</version>
  <packaging>pom</packaging>
  <name>Stratio Spark Mongodb Datasource</name>
  <description>A Spark SQL library for MongoDB</description>

spark-mongodb-examples/pom.xml

@@ -40,7 +40,7 @@
 
  <properties>
  <scala.binary.version>2.10</scala.binary.version>
- <mongodb.datasource.version>0.11.0-SNAPSHOT</mongodb.datasource.version>
+ <mongodb.datasource.version>0.11.2-RC1-SNAPSHOT</mongodb.datasource.version>
  <spark.version>1.5.2</spark.version>
  </properties>
 

spark-mongodb/src/main/scala/com/stratio/datasource/mongodb/client/MongodbClientFactory.scala

@@ -26,6 +26,7 @@ import com.mongodb.casbah.Imports._
 import com.mongodb.casbah.MongoClient
 import com.stratio.datasource.mongodb.client.MongodbClientActor._
 import com.stratio.datasource.mongodb.config.MongodbSSLOptions
+import com.typesafe.config.ConfigFactory
 
 import scala.concurrent.Await
 import scala.concurrent.duration._
@@ -43,7 +44,7 @@ object MongodbClientFactory {
  /**
  * Scheduler that close connections automatically when the timeout was expired
  */
- private val actorSystem = ActorSystem()
+ private val actorSystem = ActorSystem("mongodbClientFactory", ConfigFactory.load(ConfigFactory.parseString("akka.daemonic=on")))
  private val scheduler = actorSystem.scheduler
  private val SecondsToCheckConnections = 60
  private val mongoConnectionsActor = actorSystem.actorOf(Props(new MongodbClientActor), "mongoConnectionActor")
@@ -154,8 +155,9 @@ object MongodbClientFactory {
  }
  }
 
+ // TODO Review when refactoring config
  def extractValue[T](options: Map[String, Any], key: String): Option[T] =
- options.get(key).map(_.asInstanceOf[T])
+ options.get(key.toLowerCase).map(_.asInstanceOf[T])
 
  def sslBuilder(optionSSLOptions: Option[MongodbSSLOptions]): Boolean =
  optionSSLOptions.exists(sslOptions => {

spark-mongodb/src/main/scala/com/stratio/datasource/mongodb/config/MongodbConfig.scala

@@ -23,6 +23,7 @@ import com.stratio.datasource.util.Config._
 /**
  * Values and Functions for access and parse the configuration parameters
  */
+// TODO Review when refactoring config
 object MongodbConfig {
 
  // Parameter names
@@ -41,12 +42,15 @@ object MongodbConfig {
  val SamplingRatio = "schema_samplingRatio"
  val SplitSize = "splitSize"
  val SplitKey = "splitKey"
+ val SplitKeyType = "splitKeyType"
+ val SplitKeyMin = "splitKeyMin"
+ val SplitKeyMax = "splitKeyMax"
  val UpdateFields = "updateFields"
  val Language = "language"
  val ConnectionsTime = "connectionsTime"
  val CursorBatchSize = "cursorBatchSize"
  val BulkBatchSize = "bulkBatchSize"
- val IdAsObjectId = "idasobjectid"
+ val IdAsObjectId = "idAsObjectId"
 
  // List of parameters for mongoClientOptions
  val ListMongoClientOptions = List(
@@ -90,6 +94,7 @@ object MongodbConfig {
  * @param parameters List of parameters
  * @return List of parameters parsed to correct mongoDb configurations
  */
+ // TODO Review when refactoring config
  def parseParameters(parameters : Map[String,String]): Map[String, Any] = {
 
  // required properties
@@ -134,6 +139,7 @@ object MongodbConfig {
  * @param readPreference string key for identify the correct object
  * @return readPreference object
  */
+ // TODO Review when refactoring config
  def parseReadPreference(readPreference: String): ReadPreference = {
  readPreference.toUpperCase match {
  case "PRIMARY" => com.mongodb.casbah.ReadPreference.Primary
@@ -150,18 +156,16 @@ object MongodbConfig {
  * @param writeConcern string key for identify the correct object
  * @return writeConcern object
  */
+ // TODO Review when refactoring config
  def parseWriteConcern(writeConcern: String): WriteConcern = {
  writeConcern.toUpperCase match {
- case "FSYNC_SAFE" => com.mongodb.WriteConcern.FSYNC_SAFE
- case "FSYNCED" => com.mongodb.WriteConcern.FSYNCED
- case "JOURNAL_SAFE" => com.mongodb.WriteConcern.JOURNAL_SAFE
- case "JOURNALED" => com.mongodb.WriteConcern.JOURNALED
+ case "SAFE" | "ACKNOWLEDGED" => com.mongodb.WriteConcern.SAFE
+ case "NORMAL" | "UNACKNOWLEDGED" => com.mongodb.WriteConcern.NORMAL
+ case "REPLICAS_SAFE" | "REPLICA_ACKNOWLEDGED" => com.mongodb.WriteConcern.REPLICAS_SAFE
+ case "FSYNC_SAFE" | "FSYNCED" => com.mongodb.WriteConcern.FSYNC_SAFE
  case "MAJORITY" => com.mongodb.WriteConcern.MAJORITY
- case "NORMAL" => com.mongodb.WriteConcern.NORMAL
- case "REPLICA_ACKNOWLEDGED" => com.mongodb.WriteConcern.REPLICA_ACKNOWLEDGED
- case "REPLICAS_SAFE" => com.mongodb.WriteConcern.REPLICAS_SAFE
- case "SAFE" => com.mongodb.WriteConcern.SAFE
- case "UNACKNOWLEDGED" => com.mongodb.WriteConcern.UNACKNOWLEDGED
+ case "JOURNAL_SAFE" | "JOURNALED" => com.mongodb.WriteConcern.JOURNAL_SAFE
+ case "NONE" | "ERRORS_IGNORED" => com.mongodb.WriteConcern.NONE
  case _ => DefaultWriteConcern
  }
  }