title: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;;\nJoin Inner
body: <p>I wanna join 2 streams but I received the next error and I don't know how to fix it:</p>
  
  <blockquote>
    <p>Append output mode not supported when there are streaming aggregations
    on streaming DataFrames/DataSets without watermark;;\nJoin Inner</p>
  </blockquote>
  
  <pre><code>df_stream = spark.readStream.schema(schema_clicks).option("ignoreChanges", True).option("header", True).format("csv").load("s3://mybucket/*.csv")
  display(df_stream.select("SendID", "EventType", "EventDate"))
  </code></pre>
  
  <p><a href="https://i.stack.imgur.com/jUixl.png" rel="noreferrer"><img src="https://i.stack.imgur.com/jUixl.png" alt="enter image description here"></a></p>
  
  <p>I wanna join df1 with df2:</p>
  
  <pre><code>df1 = df_stream \
                .withColumn('timestamp', unix_timestamp(col('EventDate'), "MM/dd/yyyy hh:mm:ss aa").cast(TimestampType())) \
                .select(col("SendID"), col("timestamp"), col("EventType")) \
                .withColumnRenamed("SendID", "SendID_update") \
                .withColumnRenamed("timestamp", "timestamp_update") \
                .withWatermark("timestamp_update", "1 minutes")
  
  df2 = df_stream \
                .withColumn('timestamp', unix_timestamp(col('EventDate'), "MM/dd/yyyy hh:mm:ss aa").cast(TimestampType())) \
                .withWatermark("timestamp", "1 minutes") \
                .groupBy(col("SendID")) \
                .agg(max(col('timestamp')).alias("timestamp")) \
                .orderBy('timestamp', ascending=False)
  
  join = df2.alias("A").join(df1.alias("B"),  expr(
        "A.SendID = B.SendID_update" +
          " AND " +
          "B.timestamp_update &gt;= A.timestamp " +
          " AND " +
          "B.timestamp_update &lt;= A.timestamp + interval 1 hour"))
  </code></pre>
  
  <p>Then finally when I write the result in append mode:</p>
  
  <pre><code>join \
  .writeStream \
  .outputMode("Append") \
  .option("checkpointLocation", "s3://checkpointjoin_delta")  \
  .format("delta")  \
  .table("test_join")
  </code></pre>
  
  <p>I received the previous error.</p>
  
  <blockquote>
    <p>AnalysisException                         Traceback (most recent call
    last)  in ()
    ----> 1 join.writeStream.outputMode("Append").option("checkpointLocation",
    "s3://checkpointjoin_delta").format("delta").table("test_join")</p>
    
    <p>/databricks/spark/python/pyspark/sql/streaming.py in table(self,
    tableName)    1137         """    1138         if
    isinstance(tableName, basestring):
    -> 1139             return self._sq(self._jwrite.table(tableName))    1140         else:    1141             raise TypeError("tableName can
    be only a single string")</p>
    
    <p>/databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py
    in <strong>call</strong>(self, *args)    1255         answer =
    self.gateway_client.send_command(command)    1256         return_value
    = get_return_value(
    -> 1257             answer, self.gateway_client, self.target_id, self.name)    1258     1259         for temp_arg in temp_args:</p>
    
    <p>/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
         67                                              e.java_exception.getStackTrace()))</p>
  </blockquote>