title: SparkStreaming: avoid checkpointLocation check
body: <p>I'm writing a library to integrate Apache Spark with a custom environment. I'm implementing both custom streaming sources and streaming writers.</p>
  
  <p>Some of the sources I'm developing are not recoverable, at least after a crash of the application. If an application is restarted, it needs to reload all the data.
  Therefore we would like to avoid the users to have to set explicitly the 'checkpointLocation' option.
  But if the option is not provided, we see the following error:</p>
  
  <pre><code>org.apache.spark.sql.AnalysisException: checkpointLocation must be specified either through option("checkpointLocation", ...) or SparkSession.conf.set("spark.sql.streaming.checkpointLocation", ...);
  </code></pre>
  
  <p>However, if I use the console stream output, everything works fine.</p>
  
  <p>Is there a way to obtain the same behavior?</p>
  
  <p>Note: we are using the Spark v2 interfaces for our stream readers/writers.</p>
  
  <hr>
  
  <p>Spark log:</p>
  
  <pre><code>18/06/29 16:36:48 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/C:/mydir/spark-warehouse/').
  18/06/29 16:36:48 INFO SharedState: Warehouse path is 'file:/C:/mydir/spark-warehouse/'.
  18/06/29 16:36:48 INFO StateStoreCoordinatorRef: Registered StateStoreCoordinator endpoint
  org.apache.spark.sql.AnalysisException: checkpointLocation must be specified either through option("checkpointLocation", ...) or SparkSession.conf.set("spark.sql.streaming.checkpointLocation", ...);
      at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$3.apply(StreamingQueryManager.scala:213)
      at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$3.apply(StreamingQueryManager.scala:208)
      at scala.Option.getOrElse(Option.scala:121)
      at org.apache.spark.sql.streaming.StreamingQueryManager.createQuery(StreamingQueryManager.scala:207)
      at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:299)
      at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:296)
      ...
  18/06/29 16:36:50 INFO SparkContext: Invoking stop() from shutdown hook
  </code></pre>
  
  <p>This is how I start the streaming job:</p>
  
  <pre><code>spark.readStream().format("mysource").load()
    .writeStream().format("mywriter").outputMode(OutputMode.Append()).start();
  </code></pre>
  
  <p>Everything works fine, instead, if for example I run:</p>
  
  <pre><code>spark.readStream().format("mysource").load()
    .writeStream().format("console").outputMode(OutputMode.Append()).start();
  </code></pre>
  
  <hr>
  
  <p>I cannot share the full code of the data writer. Anyway, I did something like this:</p>
  
  <pre><code>class MySourceProvider extends DataSourceRegister with StreamWriteSupport {
    def createStreamWriter(queryId: String, schema: StructType, mode: OutputMode, options: DataSourceOptions): StreamWriter = {
      new MyStreamWriter(...)
    }
    def shortName(): String = {
      "mywriter"
    }
  }
  
  class MyStreamWriter(...) extends StreamWriter { 
    def abort(epochId: Long, messages: Array[WriterCommitMessage]): Unit = {}
    def commit(epochId: Long, messages: Array[WriterCommitMessage]): Unit = {}
    def createWriterFactory(): DataWriterFactory[Row] = {
      new MyDataWriterFactory()
    }
  }
  </code></pre>