fixes #122 improved Spark documentation
authorKeith Turner <kturner@apache.org>
Mon, 9 Jan 2017 15:26:10 +0000 (10:26 -0500)
committerKeith Turner <kturner@apache.org>
Mon, 9 Jan 2017 15:26:10 +0000 (10:26 -0500)
README.md
docs/spark.md [new file with mode: 0644]
modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
modules/spark/src/main/java/org/apache/fluo/recipes/spark/FluoSparkHelper.java

index 27f6132..fe511a6 100644 (file)
--- a/README.md
+++ b/README.md
@@ -43,6 +43,7 @@ Recipes have common needs that are broken down into the following reusable compo
 * [Serialization][serialization] - Common code for serializing POJOs. 
 * [Transient Ranges][transient] - Standardized process for dealing with transient data.
 * [Table optimization][optimization] - Standardized process for optimizing the Fluo table.
+* [Spark integration][spark] - Spark+Fluo integration code.
 
 ### Usage
 
@@ -108,6 +109,7 @@ Below is a sample Maven POM containing all possible Fluo Recipes dependencies:
 [transient]: docs/transient.md
 [optimization]: docs/table-optimization.md
 [row-hasher]: docs/row-hasher.md
+[spark]: docs/spark.md
 [testing]: docs/testing.md
 [ti]: https://travis-ci.org/apache/incubator-fluo-recipes.svg?branch=master
 [tl]: https://travis-ci.org/apache/incubator-fluo-recipes
diff --git a/docs/spark.md b/docs/spark.md
new file mode 100644 (file)
index 0000000..197c189
--- /dev/null
@@ -0,0 +1,32 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+# Apache Spark helper code
+
+Fluo Recipes has some helper code for [Apache Spark][spark].  Most of the helper code is currently
+related to bulk importing data into Accumulo.  This is useful for initializing a new Fluo table with
+historical data via Spark.  The Spark helper code is found at
+[modules/spark/src/main/java/org/apache/fluo/recipes/spark/][sdir].
+
+For information on using Spark to load data into Fluo, check out this [blog post][blog].
+
+If you know of other Spark+Fluo integration code that would be useful, then please consider [opening
+an issue](https://github.com/apache/fluo-recipes/issues/new).
+
+[spark]: https://spark.apache.org
+[sdir]: ../modules/spark/src/main/java/org/apache/fluo/recipes/spark/
+[blog]: https://fluo.apache.org/blog/2016/12/22/spark-load/
+
index 715d330..939e9ed 100644 (file)
@@ -399,11 +399,10 @@ public class CollisionFreeMap<K, V> {
   }
 
   /**
-   * A @link {@link CollisionFreeMap} stores data in its own data format in the Fluo table. When
+   * A {@link CollisionFreeMap} stores data in its own data format in the Fluo table. When
    * initializing a Fluo table with something like Map Reduce or Spark, data will need to be written
    * in this format. Thats the purpose of this method, it provide a simple class that can do this
    * conversion.
-   *
    */
   public static <K2, V2> Initializer<K2, V2> getInitializer(String mapId, int numBuckets,
       SimpleSerializer serializer) {
index dc467b3..587f28c 100644 (file)
@@ -120,7 +120,7 @@ public class FluoSparkHelper {
   }
 
   /**
-   * Reads all data in Fluo and returns it as a RowColumn/Value RDD
+   * Reads all data from a snapshot in Fluo and returns it as a RowColumn/Value RDD.
    *
    * @param ctx Java Spark context
    * @return RowColumn/Value RDD containing all data in Fluo
@@ -139,7 +139,10 @@ public class FluoSparkHelper {
   }
 
   /**
-   * Bulk import RowColumn/Value data into Fluo
+   * Bulk import RowColumn/Value data into Fluo table (obtained from Fluo configuration). This
+   * method will repartition RDD using the current split points of the Fluo table, creating one
+   * partition per tablet in the table. This is done so that one RFile is created per tablet for
+   * bulk import.
    *
    * @param data RowColumn/Value data to import
    * @param opts Bulk import options
@@ -163,7 +166,9 @@ public class FluoSparkHelper {
   }
 
   /**
-   * Bulk import Key/Value data into Fluo
+   * Bulk import Key/Value data into into Fluo table (obtained from Fluo configuration). This method
+   * does not repartition data. One RFile will be created for each partition in the passed in RDD.
+   * Ensure the RDD is reasonably partitioned before calling this method.
    *
    * @param data Key/Value data to import
    * @param opts Bulk import options
@@ -173,7 +178,9 @@ public class FluoSparkHelper {
   }
 
   /**
-   * Bulk import RowColumn/Value data into Accumulo
+   * Bulk import RowColumn/Value data into specified Accumulo table. This method will repartition
+   * RDD using the current split points of the specified table, creating one partition per tablet in
+   * the table. This is done so that one RFile is created per tablet for bulk import.
    *
    * @param data RowColumn/Value data to import
    * @param accumuloTable Accumulo table used for import
@@ -196,7 +203,9 @@ public class FluoSparkHelper {
   }
 
   /**
-   * Bulk import Key/Value data into Accumulo
+   * Bulk import Key/Value data into specified Accumulo table. This method does not repartition
+   * data. One RFile will be created for each partition in the passed in RDD. Ensure the RDD is
+   * reasonably partitioned before calling this method.
    *
    * @param data Key/value data to import
    * @param accumuloTable Accumulo table used for import
@@ -262,12 +271,27 @@ public class FluoSparkHelper {
 
     public BulkImportOptions() {}
 
+    /**
+     * If this methods is not called, then a Connector will be created using properties in the
+     * FluoConfiguration supplied to
+     * {@link FluoSparkHelper#FluoSparkHelper(FluoConfiguration, Configuration, Path)}
+     * 
+     * @param conn Use this connector to bulk import files into Accumulo.
+     * @return this
+     */
     public BulkImportOptions setAccumuloConnector(Connector conn) {
       Objects.requireNonNull(conn);
       this.conn = conn;
       return this;
     }
 
+    /**
+     * If this method is not called, then a temp dir will be created based on the path passed
+     * supplied to {@link FluoSparkHelper#FluoSparkHelper(FluoConfiguration, Configuration, Path)}
+     * 
+     * @param tempDir Use this directory to store RFiles generated for bulk import.
+     * @return this
+     */
     public BulkImportOptions setTempDir(Path tempDir) {
       Objects.requireNonNull(tempDir);
       this.tempDir = tempDir;