A suggestion for Monadizing Map Algebra operations, so that they're generic across any Tile type.

MapAlgebra.scala

 /* Eugene (et al.), our mistake might have been that `MapAlgebra` isn't the Monad,                                                                                                                                 
  * RDDs are. We'd like to do things like:                                                                                                                                                                          
  *   rdd1.localMean(rdd2).flatMap(r => r.focalGradient).flatMap(r => r.moreStuff)                                                                                                                                  
  */

 implicit class KeyedRDDMethods[K: Key, A](rdd: RDD[(K,A)]) {
   /* Binary operations are naturally expressed as an FP `zipWith` operation.                                                                                                                                       
    * There is a keyed variant that works across Map-like containers.                                                                                                                                         
    */
   def zipWithKey(other: RDD[(K,B)])(f: (K,A,B) => C): RDD[(K,C)] = {
     /* For (Key,Tile) pairs who have a matching Key in the other RDD,                                                                                                                                              
      * combine their Tiles via some `f`, automatically retaining their
      * shared Key.
      */
     ???
   }

 }

 implicit class MapAlgebraMethods[K: Key, A](rdd: RDD[(K,A)]) {
   /* With the above, we can implement Map Algebra functions in the following                                                                                                                                       
    * style, where the optional `g` function is the means by which the user                                                                                                                                         
    * can "lift" the `Tile` output of the algebra back into an `A` or otherwise.                                                                                                                                            
    * Usage then looks like:                                                                                                                                                                                        
    *                                                                                                                                                                                                               
    *   rdd.localMean(other)  // "I'm okay with dumping metadata." Type: RDD[(Key, Tile)]                                                                                                                           
    *   rdd.localMean(other) { case (a1,a2,t) =>                                                                                                                                                                    
    *     HigherTileType(t, a.metadata)  // "Dump a2's metadata" Type: RDD[(Key, HigherTileType)]                                                                                                                   
    *   }                                                                                                                                                                                                           
    *                                                                                                                                                                                                               
    * This pattern leaves RDDs flatMappable for subsequent MapAlgebra ops.                                                                                                                                          
    */
   def localMean(other: RDD[(K,A)])(g: (A,A,B) => C = _._3): RDD[(K, C)] = {

     rdd.zipWithKey(other)({ case (_,a,b) =>
       val mean: Tile = doTileMean(a.tile, b.tile)
       g(a,b,mean)
     })

   }
 }