<jupyter_start><jupyter_text>d-sandbox   -sandbox Working with Time Windows In this lesson you:* Use sliding windows to aggregate over chunks of data rather than all data* Apply watermarking to throw away stale old data that you do not have space to keep* Plot live graphs using `display` Audience* Primary Audience: Data Engineers* Secondary Audience: Data Scientists, Software Engineers Prerequisites* Web browser: current versions of Google Chrome, Firefox, Safari, Microsoft Edge and Internet Explorer 11 on Windows 7, 8, or 10 (see Supported Web Browsers)* Databricks Runtime 4.2 or greater* Completed courses DataFrames, ETL-Parts 1, 2 and 3 from Databricks Academy, or have similar knowledge <iframe  src="//fast.wistia.net/embed/iframe/uiwie12hng?videoFoam=true"style="border:1px solid 1cb1c2;"allowtransparency="true" scrolling="no" class="wistia_embed"name="wistia_embed" allowfullscreen mozallowfullscreen webkitallowfullscreenoallowfullscreen msallowfullscreen width="640" height="360" >  &nbsp;Watch full-screen.  Getting StartedRun the following cell to configure our "classroom."<jupyter_code>%run "./Includes/Classroom-Setup"<jupyter_output><empty_output><jupyter_text>Streaming AggregationsContinuous applications often require near real-time decisions on real-time, aggregated statistics.Some examples include * Aggregating errors in data from IoT devices by type * Detecting anomalous behavior in a server's log file by aggregating by country. * Doing behavior analysis on instant messages via hash tags.However, in the case of streams, you generally don't want to run aggregations over the entire dataset.  What problems might you encounter if you aggregate over a stream's entire dataset? -sandbox  window.onload = function() {    var allHints = document.getElementsByClassName("hint-1068");    var answer = document.getElementById("answer-1068");    var totalHints = allHints.length;    var nextHint = 0;    var hasAnswer = (answer != null);    var items = new Array();    var answerLabel = "Click here for the answer";    for (var i = 0; i < totalHints; i++) {      var elem = allHints[i];      var label = "";      if ((i + 1) == totalHints)        label = answerLabel;      else        label = "Click here for the next hint";      items.push({label: label, elem: elem});    }    if (hasAnswer) {      items.push({label: '', elem: answer});    }    var button = document.getElementById("hint-button-1068");    if (totalHints == 0) {      button.innerHTML = answerLabel;    }    button.onclick = function() {      items[nextHint].elem.style.display = 'block';      if ((nextHint + 1) >= items.length)        button.style.display = 'none';      else        button.innerHTML = items[nextHint].label;        nextHint += 1;    };    button.ondblclick = function(e) {      e.stopPropagation();    }    var answerCodeBlocks = document.getElementsByTagName("code");    for (var i = 0; i < answerCodeBlocks.length; i++) {      var elem = answerCodeBlocks[i];      var parent = elem.parentNode;      if (parent.name != "pre") {        var newNode = document.createElement("pre");        newNode.append(elem.cloneNode(true));        elem.replaceWith(newNode);        elem = newNode;      }      elem.ondblclick = function(e) {        e.stopPropagation();      };      elem.style.marginTop = "1em";    }  };  <button type="button" class="btn btn-light"          style="margin-top: 1em"          id="hint-button-1068">Click here for a hint  The answer:  While streams have a definitive start, there conceptually is no end to the flow of data.Because there is no &quot;end&quot; to a stream, the size of the dataset grows in perpetuity.This means that your cluster will eventually run out of resources.Instead of aggregating over the entire dataset, you can aggregate over data grouped by windows of time (say, every 5 minutes or every hour).This is referred to as windowing    WindowingIf we were using a static DataFrame to produce an aggregate count, we could use `groupBy()` and `count()`. Instead we accumulate counts within a sliding window, answering questions like "How many records are we getting every second?"The following illustration, from the Structured Streaming Programming Guide guide, helps us understanding how it works:   Event Time vs Receipt Time**Event Time** is the time at which the event occurred in the real world.**Event Time** is **NOT** something maintained by the Structured Streaming framework. At best, Streams only knows about **Receipt Time** - the time a piece of data arrived in Spark.  What are some examples of **Event Time**? **of Receipt Time**? -sandbox  window.onload = function() {    var allHints = document.getElementsByClassName("hint-2254");    var answer = document.getElementById("answer-2254");    var totalHints = allHints.length;    var nextHint = 0;    var hasAnswer = (answer != null);    var items = new Array();    var answerLabel = "Click here for the answer";    for (var i = 0; i < totalHints; i++) {      var elem = allHints[i];      var label = "";      if ((i + 1) == totalHints)        label = answerLabel;      else        label = "Click here for the next hint";      items.push({label: label, elem: elem});    }    if (hasAnswer) {      items.push({label: '', elem: answer});    }    var button = document.getElementById("hint-button-2254");    if (totalHints == 0) {      button.innerHTML = answerLabel;    }    button.onclick = function() {      items[nextHint].elem.style.display = 'block';      if ((nextHint + 1) >= items.length)        button.style.display = 'none';      else        button.innerHTML = items[nextHint].label;        nextHint += 1;    };    button.ondblclick = function(e) {      e.stopPropagation();    }    var answerCodeBlocks = document.getElementsByTagName("code");    for (var i = 0; i < answerCodeBlocks.length; i++) {      var elem = answerCodeBlocks[i];      var parent = elem.parentNode;      if (parent.name != "pre") {        var newNode = document.createElement("pre");        newNode.append(elem.cloneNode(true));        elem.replaceWith(newNode);        elem = newNode;      }      elem.ondblclick = function(e) {        e.stopPropagation();      };      elem.style.marginTop = "1em";    }  };  <button type="button" class="btn btn-light"          style="margin-top: 1em"          id="hint-button-2254">Click here for a hint  The answer:   Examples of *Event Time*:* The timestamp recorded in each record of a log file* The instant at which an IoT device took a measurement* The moment a REST API received a request Examples of *Receipt Time*:* A timestamp added to a DataFrame the moment it was processed by Spark* The timestamp extracted from an hourly log file&x27;s file name* The time at which an IoT hub received a report of a device&x27;s measurement   - Presumably offset by some delay from when the measurement was taken    What are some of the inherent problems with using **Receipt Time**? -sandbox  window.onload = function() {    var allHints = document.getElementsByClassName("hint-5511");    var answer = document.getElementById("answer-5511");    var totalHints = allHints.length;    var nextHint = 0;    var hasAnswer = (answer != null);    var items = new Array();    var answerLabel = "Click here for the answer";    for (var i = 0; i < totalHints; i++) {      var elem = allHints[i];      var label = "";      if ((i + 1) == totalHints)        label = answerLabel;      else        label = "Click here for the next hint";      items.push({label: label, elem: elem});    }    if (hasAnswer) {      items.push({label: '', elem: answer});    }    var button = document.getElementById("hint-button-5511");    if (totalHints == 0) {      button.innerHTML = answerLabel;    }    button.onclick = function() {      items[nextHint].elem.style.display = 'block';      if ((nextHint + 1) >= items.length)        button.style.display = 'none';      else        button.innerHTML = items[nextHint].label;        nextHint += 1;    };    button.ondblclick = function(e) {      e.stopPropagation();    }    var answerCodeBlocks = document.getElementsByTagName("code");    for (var i = 0; i < answerCodeBlocks.length; i++) {      var elem = answerCodeBlocks[i];      var parent = elem.parentNode;      if (parent.name != "pre") {        var newNode = document.createElement("pre");        newNode.append(elem.cloneNode(true));        elem.replaceWith(newNode);        elem = newNode;      }      elem.ondblclick = function(e) {        e.stopPropagation();      };      elem.style.marginTop = "1em";    }  };  <button type="button" class="btn btn-light"          style="margin-top: 1em"          id="hint-button-5511">Click here for a hint  The answer:  The main problem with using **Receipt Time** is going to be with accuracy. For example:* The time between when an IoT device takes a measurement vs when it is reported can be off by several minutes.   - This could have significant ramifications to security and health devices, for example* The timestamp embedded in an hourly log file can be off by up to one hour making correlations to other events extremely difficult* The timestamp added by Spark as part of a DataFrame transformation can be off by hours to weeks to months depending on when the event occurred and when the job ran    When might it be OK to use **Receipt Time** instead of **Event Time**? -sandbox  window.onload = function() {    var allHints = document.getElementsByClassName("hint-1624");    var answer = document.getElementById("answer-1624");    var totalHints = allHints.length;    var nextHint = 0;    var hasAnswer = (answer != null);    var items = new Array();    var answerLabel = "Click here for the answer";    for (var i = 0; i < totalHints; i++) {      var elem = allHints[i];      var label = "";      if ((i + 1) == totalHints)        label = answerLabel;      else        label = "Click here for the next hint";      items.push({label: label, elem: elem});    }    if (hasAnswer) {      items.push({label: '', elem: answer});    }    var button = document.getElementById("hint-button-1624");    if (totalHints == 0) {      button.innerHTML = answerLabel;    }    button.onclick = function() {      items[nextHint].elem.style.display = 'block';      if ((nextHint + 1) >= items.length)        button.style.display = 'none';      else        button.innerHTML = items[nextHint].label;        nextHint += 1;    };    button.ondblclick = function(e) {      e.stopPropagation();    }    var answerCodeBlocks = document.getElementsByTagName("code");    for (var i = 0; i < answerCodeBlocks.length; i++) {      var elem = answerCodeBlocks[i];      var parent = elem.parentNode;      if (parent.name != "pre") {        var newNode = document.createElement("pre");        newNode.append(elem.cloneNode(true));        elem.replaceWith(newNode);        elem = newNode;      }      elem.ondblclick = function(e) {        e.stopPropagation();      };      elem.style.marginTop = "1em";    }  };  <button type="button" class="btn btn-light"          style="margin-top: 1em"          id="hint-button-1624">Click here for a hint  The answer:  When accuracy is not a significant concern - that is **Receipt Time** is close enough to **Event Time**One example would be for IoT events that can be delayed by minutes but the resolution of your query is by days or months (close enough)    Windowed Streaming ExampleFor this example, we will examine the files in `/mnt/training/sensor-data/accelerometer/time-series-stream.json/`.Each line in the file contains a JSON record with two fields: `time` and `action`New files are being written to this directory continuously (aka streaming).Theoretically, there is no end to this process.Let's start by looking at the head of one such file:<jupyter_code>%fs head dbfs:/mnt/training/sensor-data/accelerometer/time-series-stream.json/file-0.json<jupyter_output><empty_output><jupyter_text>-sandboxLet's try to analyze these files interactively. First configure a schema. The schema must be specified for file-based Structured Streams. Because of the simplicity of the schema, we can use the simpler, DDL-formatted, string representation of the schema.<jupyter_code>inputPath = "dbfs:/mnt/training/sensor-data/accelerometer/time-series-stream.json/"

jsonSchema = "time timestamp, action string"<jupyter_output><empty_output><jupyter_text>With the schema defined, we can create the initial DataFrame `inputDf` and then `countsDF` which represents our aggregation:<jupyter_code>from pyspark.sql.functions import window, col

inputDF = (spark
  .readStream                                 # Returns an instance of DataStreamReader
  .schema(jsonSchema)                         # Set the schema of the JSON data
  .option("maxFilesPerTrigger", 1)            # Treat a sequence of files as a stream, one file at a time
  .json(inputPath)                            # Specifies the format, path and returns a DataFrame
)

#TODO
countsDF = ( inputDF
.groupBy(col("action"),# Aggregate by action...
 window(col("time"),"1 hour"))# ...then by a 1 hour window
.count()# For the aggregate, produce a count
.select(col("window.start").alias("start"),col("count"),col("action"))# Elevate field to column # Include count # Include action
.orderBy(col("start"))# Sort by the start time
)<jupyter_output><empty_output><jupyter_text>To view the results of our query, pass the DataFrame `countsDF` to the `display()` function.<jupyter_code>display(countsDF)<jupyter_output><empty_output><jupyter_text>Performance ConsiderationsIf you run that query, as is, it will take a surprisingly long time to start generating data. What's the cause of the delay? If you expand the **Spark Jobs** component, you'll see something like this:It's our `groupBy()`. `groupBy()` causes a _shuffle_, and, by default, Spark SQL shuffles to 200 partitions. In addition, we're doing a _stateful_ aggregation: one that requires Structured Streaming to maintain and aggregate data over time.When doing a stateful aggregation, Structured Streaming must maintain an in-memory _state map_ for each window within each partition. For fault tolerance reasons, the state map has to be saved after a partition is processed, and it needs to be saved somewhere fault-tolerant. To meet those requirements, the Streaming API saves the maps to a distributed store. On some clusters, that will be HDFS. Databricks uses the DBFS.That means that every time it finishes processing a window, the Streaming API writes its internal map to disk. The write has some overhead, typically between 1 and 2 seconds. One way to reduce this overhead is to reduce the number of partitions Spark shuffles to.In most cases, you want a 1-to-1 mapping of partitions to cores for streaming applications. Rerun the query below and notice the performance improvement.Once the data is loaded, render a line graph with * **Keys** is set to `start`* **Series groupings** is set to `action`* **Values** is set to `count`<jupyter_code>spark.conf.set("spark.sql.shuffle.partitions", sc.defaultParallelism)

display(countsDF)<jupyter_output><empty_output><jupyter_text>Wait until stream is done initializing...<jupyter_code>untilStreamIsReady("display_query_11")<jupyter_output><empty_output><jupyter_text>When you are done, stop all the streaming jobs.<jupyter_code>for s in spark.streams.active: # Iterate over all active streams
  s.stop()                    # Stop the stream<jupyter_output><empty_output><jupyter_text>Problem with Generating Many WindowsWe are generating a window for every 1 hour aggregate. _Every window_ has to be separately persisted and maintained.Over time, this aggregated data will build up in the driver.The end result being a massive slowdown if not an OOM Error. How do we fix that problem? -sandbox  window.onload = function() {    var allHints = document.getElementsByClassName("hint-5927");    var answer = document.getElementById("answer-5927");    var totalHints = allHints.length;    var nextHint = 0;    var hasAnswer = (answer != null);    var items = new Array();    var answerLabel = "Click here for the answer";    for (var i = 0; i < totalHints; i++) {      var elem = allHints[i];      var label = "";      if ((i + 1) == totalHints)        label = answerLabel;      else        label = "Click here for the next hint";      items.push({label: label, elem: elem});    }    if (hasAnswer) {      items.push({label: '', elem: answer});    }    var button = document.getElementById("hint-button-5927");    if (totalHints == 0) {      button.innerHTML = answerLabel;    }    button.onclick = function() {      items[nextHint].elem.style.display = 'block';      if ((nextHint + 1) >= items.length)        button.style.display = 'none';      else        button.innerHTML = items[nextHint].label;        nextHint += 1;    };    button.ondblclick = function(e) {      e.stopPropagation();    }    var answerCodeBlocks = document.getElementsByTagName("code");    for (var i = 0; i < answerCodeBlocks.length; i++) {      var elem = answerCodeBlocks[i];      var parent = elem.parentNode;      if (parent.name != "pre") {        var newNode = document.createElement("pre");        newNode.append(elem.cloneNode(true));        elem.replaceWith(newNode);        elem = newNode;      }      elem.ondblclick = function(e) {        e.stopPropagation();      };      elem.style.marginTop = "1em";    }  };  <button type="button" class="btn btn-light"          style="margin-top: 1em"          id="hint-button-5927">Click here for a hint  The answer:  One simple solution is to increase the size of our window (say, to 2 hours).That way, we&x27;re generating fewer windows.But if the job runs for a long time, we&x27;re still building up an unbounded set of windows.Eventually, we could hit resource limits.    WatermarkingA better solution to the problem is to define a cut-off.A point after which Structured Streaming is allowed to throw saved windows away.That's what _watermarking_ allows us to do.  Refining our previous exampleBelow is our previous example with watermarking. We're telling Structured Streaming to keep no more than 2 hours of aggregated data.<jupyter_code>watermarkedDF = (inputDF
.withWatermark("time","2 hours")#Specify a 2-hour watermark
  .groupBy(col("action"),                     # Aggregate by action...
           window(col("time"), "1 hour"))     # ...then by a 1 hour window
  .count()                                    # For each aggregate, produce a count
  .select(col("window.start").alias("start"), # Elevate field to column
          col("count"),                       # Include count
          col("action"))                      # Include action
  .orderBy(col("start"))                      # Sort by the start time
)
display(watermarkedDF)                        # Start the stream and display it<jupyter_output><empty_output><jupyter_text>In the example above,   * Data received 2 hour _past_ the watermark will be dropped. * Data received within 2 hours of the watermark will never be dropped.More specifically, any data less than 2 hours behind the latest data processed till then is guaranteed to be aggregated.However, the guarantee is strict only in one direction. Data delayed by more than 2 hours is not guaranteed to be dropped; it may or may not get aggregated. The more delayed the data is, the less likely the engine is going to process it. Wait until stream is done initializing...<jupyter_code>#TODO
untilStreamIsReady("display_query_12")<jupyter_output><empty_output><jupyter_text>Stop all the streams<jupyter_code>for s in spark.streams.active: # Iterate over all active streams
  s.stop()                     # Stop the stream<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10188909822292377, 'fraction_numerical': 0.009318193390702174, 'mean_word_length': 3.4983967017865325, 'pattern_counts': {'":': 0, '<': 61, '<?xml version=': 0, '>': 56, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '634957', 'license': 'MIT', 'n_tokens_mistral': 5456, 'n_tokens_neox': 5227, 'n_words': 2072, 'path': 'Structured Streaming/SS 03 - Time Windows.ipynb', 'repo_name': 'vshiv667/Data-Engineering', 'size': '19639'}