title: How to use fully formed SQL with spark structured streaming
body: <p>Documentation for Spark structured streaming says that - as of spark 2.3 all methods on the spark context available for <em>static</em> <code>DataFrame</code>/<code>DataSet</code>'s are also available for use with <em>structured streaming</em> <code>DataFrame</code>/<code>DataSet</code>'s as well.   However I have yet to run across any <em>examples</em> of same. </p>
  
  <p>Using fully formed SQL's is more flexible, expressive, and productive for me than the <code>DSL</code>. In addition for my use case those SQL's are already developed and well tested for <em>static</em> versions. There <em>must</em> be <em>some</em> rework - in particular to use <code>join</code>s in place of <code>correlated subqueries</code>. However there is still much value in retaining the overall full-bodied sql structure.</p>
  
  <p>The format for which I am looking to use is like this hypothetical join:</p>
  
  <pre><code> val tabaDf = spark.readStream(..)
   val tabbDf = spark.readStream(..)
  
   val joinSql = """select a.*, 
                    b.productName 
                    from taba
                    join tabb 
                    on a.productId = b.productId
                    where ..
                    group by ..
                    having ..
                    order by .."""
   val joinedStreamingDf = spark.sql(joinSql)
  </code></pre>
  
  <p>There are a couple of items that are not clear how to do:</p>
  
  <ul>
  <li><p>Are the <code>tabaDf</code> and <code>tabbDf</code> supposed to be defined via <code>spark.readStream</code>: this is my assumption</p></li>
  <li><p>How to declare <code>taba</code> and <code>tabb</code> . Trying to use </p>
  
  <pre><code>tabaDf.createOrReplaceTempView("taba")
  tabbDf.createOrReplaceTempView("tabb")
  </code></pre>
  
  <p>results in </p>
  
  <blockquote>
    <p>WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException</p>
  </blockquote></li>
  </ul>
  
  <p>All of the examples I could find are  using the <code>DSL</code> and/or the <code>selectExpr()</code> - like the following <a href="https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html" rel="noreferrer">https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html</a></p>
  
  <pre><code> df.selectExpr("CAST(userId AS STRING) AS key", "to_json(struct(*)) AS value")
  </code></pre>
  
  <p>or using <code>select</code> :</p>
  
  <pre><code>sightingLoc
    .groupBy("zip_code", window("start_time", "1 hour"))
    .count()
    .select( 
      to_json(struct("zip_code", "window")).alias("key"),
      col("count").cast("string").alias("value")) 
  </code></pre>
  
  <p>Are those truly the only options - so that the documentation saying that <em>all</em> methods supported on the <code>static</code> dataframe/datasets are not really accurate?   Otherwise: aAny pointers on how to correct the above issue(s) and use straight-up <code>sql</code> with streaming would be appreciated.</p>