One implication of this change is that users who want to override persist() as a no-op still need to override the required baseOutputURI even though it would be unused. And now the only reason we have the abstract baseOutputURI is to be used in the default implementation of persist(). This might be an acceptable weirdness for an uncommon use case.

Hey, Jon. Thanks for the contribution! I like the idea.

It's been a bit since I've been in this code, but it seems like perhaps the baseOutputURI could be converted to a parameter to persist() as you've specified it. It always felt a bit odd to have it as a class member, and this is probably why—saving files to some location as the only option for export is not very general. We can talk about proper signatures, but I'll suggest persist: java.net.URI → Unit for a start. In which case, VectorPipe::apply should probably take in the destination URI (which could be maybe have a null value if the user knows the pipeline has a no-op persist implementation).

Let me know what you think.

@jpolchlo Yea that idea cross my mind, but I still find a little odd that the user would need to provide the output URI to VectorPipe::apply only for it to be provided back to them in the call to Pipeline::persist when it is implemented by the user. However, this is fine when the user doesn't override the default persist implementation.
One approach that I think might be the best for both use cases is to instead remove Pipeline.persist and the call to persist from VectorPipe::apply entirely, and instead just let the user do what they want with the returned vector tiles rather than making the assumption for them. The vectorpipe.vectortile.export.saveVectorTiles() util can maybe be used by the user in the same way that vectorpipe.OSM.toGeometry() is used to prepare the input data, except here we have util to assist with what to do with the output data. The VectorPipe::apply method could also be overloaded with the additional case provided the outputUri used to invoke the default export.saveVectorTiles() which would be functionally equivalent to the current implementation except that the user provides the outputUri (like you mentioned) for cases where the default S3/HDFS output is needed immediately after the vectortiles are processed.

I'm thinking something like this:

def apply(input: DataFrame, pipeline: Pipeline, options: Options, outputUri: URI): Unit = {
  val vectorTilesByZoom = apply(input, pipeline, options)
  
  vectorTilesByZoom.foreach { case (zoom, vectorTiles) =>
    saveVectorTiles(vectorTiles, zoom, outputUri)
  }
}

def apply(input: DataFrame, pipeline: Pipeline, options: Options): Map[Int, RDD[(SpatialKey, VectorTile)]] = {
  // same impl except doesn't persist
}

Also, the overloaded apply could probably just return Unit assuming the user doesn't want to do anything else with the data after the data was written much like the current implementation.

What do you think? I have this change ready and can update this PR if it sounds good to you.

As much as it would help solve some weirdnesses to extract the write step out of the pipeline, it does kinda ruin the completeness of the pipeline, and so I'd rather consider some alternative.

It occurs to me that your addition of Pipeline::persist() is correct. This singles out the baseOutputURI as the odd member that should be killed from Pipeline. I would then suggest that it's up to the pipeline designer to ensure that the output location is communicated clearly when the pipeline is created. Sadly, this makes it harder to specify a default implementation of persist, and then implies that the baseOutputURI is required after all. And we're back to the same crappy starting point. Unless Pipeline has no default persist implementation, and a Pipeline subclass which provides the standard output routine and baseOutputURI is provided (FileSystemOutputPipeline? (ewww)).

On a second point, your amendment to VectorPipe::apply to have two forms makes sense, but I am a bit weirded out by the Map[Int, RDD[...]], and I wonder if an Iterator would confer some improved laziness? I'm concerned about having to do all the computation all at once and hold on to all past zoom levels to fill the map.

@jpolchlo Hey sorry for such a late response! I forgot about this PR, but now I'm back working with VectorPipe and I think I have a good solution which has addressed your good points.
From the commit message:

Add Pipeline.Output mixin trait and revert return type back to Unit

The Pipeline.Output mixin trait provides the existing default output behavior saving to S3 or Hadoop using the baseOutputURI which has been moved out of the base Pipeline trait. Additionally, the persist() method was renamed to finalize() to be more general since users may not want to persist immediately or even persist at all. I feel this approach provides a good abstraction while making the pipeline implementation more clear as it could then be implemented like:
case class SomePipeline(baseOutputURI: URI) extends Pipeline with Pipeline.Output

The Pipeline scaladocs were also updated and fixed to create valid links to referenced code.

The finalize method is described as

Receive the final RDD of generated vector tiles for a zoom level to finish any additional processing or handling such as persisting to a datastore.

So it does not make assumptions about what the user wants to do with the final RDD of vector tiles. In my case, I actually want to do some additional processing before persisting, so this generalization makes more sense to me. This allows separation of concern for "output" by moving the baseOutputURI to a mixin trait in the companion object enabling users to mixin the current default output behavior. Of course this would be a breaking change and so a minor version bump would probably make sense I think

Additionally, I reverted the VectorPipe.apply return type change back to Unit since the Pipeline receives the vectortiles for each zoom level anyway, so returning a Map of the RDDs is unnecessary and ugly.