Spark Struct Streaming - other operations

1. deduplication
   
   

   / Without watermark using guid column streamingDf.dropDuplicates("guid")   // With watermark using guid and eventTime columns streamingDf   .withWatermark("eventTime", "10 seconds")   .dropDuplicates("guid", "eventTime")

   watermark - watermark 까지 (참고로 이전 date 는 중복제거가 안 될 수 있음)
   without - sate 되는대까지

2. Arbitary State
   Since Spark 2.2, this can be done using the operation mapGroupsWithState and the more powerful operation flatMapGroupsWithState. Both operations allow you to apply user-defined code on grouped Datasets to update user-defined state. For more concrete details, take a look at the API documentation (Scala/Java) and the examples (Scala/Java).
   위의 Operation 은 유저가원하는(임의의) state 를 만든다. 다만 그 State 가 어떻게 활용되는지는 별이야기가없다.

     3. Unsupported!

Unsupported Operations

There are a few DataFrame/Dataset operations that are not supported with streaming DataFrames/Datasets. Some of them are as follows.

• Multiple streaming aggregations (i.e. a chain of aggregations on a streaming DF) are not yet supported on streaming Datasets.

• Limit and take first N rows are not supported on streaming Datasets.

• Distinct operations on streaming Datasets are not supported.

• Sorting operations are supported on streaming Datasets only after an aggregation and in Complete Output Mode.

• Few types of outer joins on streaming Datasets are not supported. See the support matrix in the Join Operations section for more details.

In addition, there are some Dataset methods that will not work on streaming Datasets. They are actions that will immediately run queries and return results, which does not make sense on a streaming Dataset. Rather, those functionalities can be done by explicitly starting a streaming query (see the next section regarding that).

• count() - Cannot return a single count from a streaming Dataset. Instead, use ds.groupBy().count() which returns a streaming Dataset containing a running count.

• foreach() - Instead use ds.writeStream.foreach(...) (see next section).

• show() - Instead use the console sink (see next section).

val query = df.writeStream.format("console").start()   // get the query object   query.id          // get the unique identifier of the running query that persists across restarts from checkpoint data   query.runId       // get the unique id of this run of the query, which will be generated at every start/restart   query.name        // get the name of the auto-generated or user-specified name   query.explain()   // print detailed explanations of the query   query.stop()      // stop the query   query.awaitTermination()   // block until query is terminated, with stop() or with error   query.exception       // the exception if the query has been terminated with error   query.recentProgress  // an array of the most recent progress updates for this query   query.lastProgress    // the most recent progress update of this streaming query             val spark: SparkSession = ...   spark.streams.active    // get the list of currently active streaming queries   spark.streams.get(id)   // get a query object by its unique id   spark.streams.awaitAnyTermination()   // block until any one of them terminates