<jupyter_start><jupyter_text>Twitter ExampleIn this notebook, we will work with some Twitter data. It was downloaded from *The Interet Archive* at https://archive.org/details/twitterstream. To demonstrate some use case for Spark window functions, we want to find the latest tweet for each hashtag.  1 Load Twitter DataIn a first step, we load the Twitter data. It is stored as JSONs, which are well supported by Spark.<jupyter_code>basedir = "s3://dimajix-training/data"
data = spark.read\
    .json(basedir + "/twitter-sample/00.json")<jupyter_output><empty_output><jupyter_text>Inspect SchemaNow let us inspect the schema. As we will see, the meta data for each tweet is really massive and the complete data model quite complex. Fortunately we are only interested in the tweet itself and the list of hashtags. Note that the hashtags are already extracted for us, so there is no need to use some custom extractor.<jupyter_code>data.printSchema()<jupyter_output><empty_output><jupyter_text>2 Reduce SchemaSince we don't want to work with the whole schema, let us select only the relevant columns. Note that this is only a simplification for us human beings. Spark itself would also only extract the required columns anyway, so there is no performance improvement here (which is a good thing, since Spark automatically optimizes performance).Specifically we are interested in the following columns:* `created_at` contains the date and time when the tweet was originally created* `text` contains the full text of the tweet* `entities.hashtags.text` contains an array of all hash tags<jupyter_code>hashtags = data.select(
    data["created_at"],
    data["text"],
    data["entities.hashtags.text"].alias("hashtags_array")
)
hashtags.printSchema()<jupyter_output><empty_output><jupyter_text>3 Unpack HashtagsNow the schema contains an array element with a list of all hashtags. But what we want and need is one record per hashtag with all other attributes copied into the generated records. This can be done with the Spark function `explode`. So we try again, but this time we generate a new record for every entry in the hashtag array.<jupyter_code>hashtags = data.select(
    data["created_at"],
    data["text"],
    # YOUR CODE HERE
)
hashtags.printSchema()<jupyter_output><empty_output><jupyter_text>4 Count Hashtag FrequencyOur primary goal is to find the latest tweet for every hashtag. But this only makes sense, if individual hashtags are present more than only once in our data set. So as a pre-analysis step, let us count the frequency of all hashtags.<jupyter_code>result = # YOUR CODE HERE

result.orderBy(result["count"].desc()).limit(20).toPandas()<jupyter_output><empty_output><jupyter_text>5 Find Latest Tweet per HashtagNow we want to find the newest/latest tweet for every hashtag. This could be done using a self join, but using windows is much simpler and more natural. In addition to the latest tweet, we also want to have the count of every hashtag. We already did that before, but if we want to combine both data sets, this would require a join. Instead we also count using a window function.In the first step, we simply perform the window aggregation and inspect the intermediate result<jupyter_code># First window for finding the newest hash tag. We will use the row number within the window to select the newest hash tag
rank_window = # YOUR CODE HERE

# Second window for counting the total frequency of every hash tag
count_window = # YOUR CODE HERE

ranked_hashtags = hashtags.select(
    # YOUR CODE HERE
)
ranked_hashtags.printSchema()<jupyter_output><empty_output><jupyter_text>Inspect resultNow let us inspect the intermediate result. We do not want to view all records, but we want to restrict ourselves to the non-trivial cases where there are multiple tweets for a given hashtag (i.e. `count > 1`). Moreover we also want to sort the result* First sort by count, descending. This ensures that the most commonly used hashtag comes first* Then sort by hashtag in case that there are two hashtags with the same count* Finally sort by rankThis sorting more or less gives us the windows concatenated into a new data frame.<jupyter_code>result = ranked_hashtags.filter(ranked_hashtags["count"] > 1) \
    .orderBy(ranked_hashtags["count"].desc(), ranked_hashtags["hashtag"], ranked_hashtags["rank"].asc())

result.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Find latest TweetNow we only need to filter the result and select the tweets with `rank == 1`.<jupyter_code>result = # YOUR CODE HERE

result.limit(10).toPandas()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.0665651511855558, 'fraction_numerical': 0.0036980639547531, 'mean_word_length': 5.2219215155615695, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 35, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '152656', 'license': 'MIT', 'n_tokens_mistral': 1295, 'n_tokens_neox': 1235, 'n_words': 596, 'path': 'spark-training/spark-python/jupyter-advanced-windows/Twitter Latest Tweet - Skeleton.ipynb', 'repo_name': 'kryvokhyzha/examples-and-courses', 'size': '4597'}