<jupyter_start><jupyter_text>NLP exploration with Reddit dataI came across this old article about [Creating a Data Table using Data from Reddit](https://medium.com/swlh/dashboards-in-python-using-dash-creating-a-data-table-using-data-from-reddit-1d6c0cecb4bd) and I enjoyed the idea of [Reddit](https://www.reddit.com/) data exploration. So I thought, why not try it out with atoti?That is indeed what I did. It's a pretty short notebook used to explore the data downloaded with [atoti](https://www.atoti.io/).Hence I did some additional work:- [Natural Language Processing (NLP)](https://en.wikipedia.org/wiki/Natural_language_processing) with [spaCy](https://spacy.io/) to extract [named entities](https://spacy.io/api/annotationnamed-entities)- Real-time dashboarding with atotiTo achieve real-time, I used threads to trigger the below process flow on regular intervals: Pre-requisite:You should have the Reddit [API key](https://medium.com/swlh/dashboards-in-python-using-dash-creating-a-data-table-using-data-from-reddit-1d6c0cecb4bd08db) in order to perform scraping with PRAW.  Also, uncomment and run the next 2 commands if you haven't install spaCy before.<jupyter_code># if this doesn't work, try running pip install praw in terminal
# !pip install praw spacy
# This only has to be downloaded once. Uncomment if you haven't
# !python -m spacy download en_core_web_sm
import sched
import time
from collections import Counter
from datetime import datetime

import atoti as tt
import pandas as pd
import praw
import spacy
from atoti.config import create_config<jupyter_output><empty_output><jupyter_text>NLP with spaCy and atoti  I'm going to use spaCy to perform some NLP to extract different named entities from the posting.  To do so, I have found a nice [reference](https://realpython.com/natural-language-processing-spacy-python/) article. Let's start by loading the [starter model](https://spacy.io/models) from spaCy.<jupyter_code>nlp = spacy.load("en_core_web_sm")<jupyter_output><empty_output><jupyter_text>NLP - Extracting named entities with spaCyBefore going into the implementation of the NLP, I'm creating a `entities.csv` to which I will write the entities’ attributes once I extracted them from the post.  This is in preparation for real-time data loading into atoti datacube where I could leverage on the `watch` feature of `session.read_csv` to trigger the loading on update of the file.<jupyter_code>entity_file_name = "entities.csv"
entity_file = open(entity_file_name, "w", encoding="utf-8")
entity_file.write("id|category|text|text count\n")
entity_file.flush()<jupyter_output><empty_output><jupyter_text>With spaCy, I am going to extract the following [named entities](https://spacy.io/api/annotationnamed-entities) from each Reddit posting:  - Organizations- Person- Geographical Locations- Events- Product- NORP (Nationalities/religious/political groups)In the `get_post_entity` function below, I create a processed [Doc](https://spacy.io/api/doc) object that gives a sequence of [tokens](https://spacy.io/api/token) for each post. Using the tokens, I can identify the entity with its label.  By using `Counter(list).items()`, I will be able to get a list containing all the terms related to the mentioned entity type and the number of time it appeared. Note that you could also get _n_ number of _most common_ items from the list using [Counter](https://docs.python.org/2/library/collections.html). However, I chose to get all items as I could easily use atoti's filter to achieve the same result. The list of entities with its post's id is output to the `entities.csv` created above.<jupyter_code># convert list of tuples to list of entity with its attributes
def get_entity_list(id, category, item_list):
    # perform data clean up such as trim and removing line break and double-quotes
    return list(
        map(
            lambda a: "|".join(
                '"{}"'.format(str(e).strip().replace("\n", " ").replace('"', '""'))
                for e in (
                    (
                        id,
                        category,
                    )
                    + a
                )
            ),
            list(item_list),
        )
    )


# for each posting/title, we extract the entities
def get_post_entity(id, text):

    post_entity = []
    org_list = []
    person_list = []
    gpe_list = []
    event_list = []
    product_list = []
    norp_list = []

    nlp_text = nlp(text)
    # ent is a Span object with various attributes
    for ent in nlp_text.ents:
        if ent.label_ == "ORG":
            org_list.append(ent.text)
        elif ent.label_ == "PERSON":
            person_list.append(ent.text)
        elif ent.label_ == "GPE":
            gpe_list.append(ent.text)
        elif ent.label_ == "EVENT":
            event_list.append(ent.text)
        elif ent.label_ == "PRODUCT":
            product_list.append(ent.text)
        elif ent.label_ == "NORP":
            product_list.append(ent.text)

    org = Counter(org_list).items()
    post_entity = get_entity_list(id, "Organization", org)

    person = Counter(person_list).items()
    post_entity = post_entity + get_entity_list(id, "Person", person)

    gpe = Counter(gpe_list).items()
    post_entity = post_entity + get_entity_list(id, "Geographical Location", gpe)

    event = Counter(event_list).items()
    post_entity = post_entity + get_entity_list(id, "Event", event)

    products = Counter(product_list).items()
    post_entity = post_entity + get_entity_list(id, "Product", products)

    norp = Counter(norp_list).items()
    post_entity = post_entity + get_entity_list(id, "NORP", norp)

    if len(post_entity) > 0:
        entity_file.writelines("%s\n" % x for x in post_entity)

    return post_entity<jupyter_output><empty_output><jupyter_text>NLP - Preprocessing with spaCyThrough preprocessing, I normalized the text:- lowercase- remove [stop words](https://en.wikipedia.org/wiki/Stop_words) (words that doesn't add much meaning to the sentence) and punctuation symbols- [lemmatizes](https://en.wikipedia.org/wiki/Lemmatisation) each token.This way, the named entities extracted can easily be grouped to compute the number of times it is mentioned.<jupyter_code># https://realpython.com/natural-language-processing-spacy-python/
def is_token_allowed(token):
    if not token or not token.text.strip() or token.is_stop or token.is_punct:
        return False
    return True


def preprocess_token(token):
    # Reduce token to its lowercase lemma form
    return token.lemma_.strip().lower()


def normalize(category, id, post):
    nlp_text = nlp(post)

    complete_filtered_tokens = [
        preprocess_token(token) for token in nlp_text if is_token_allowed(token)
    ]

    lemmatized_sentence = " ".join(complete_filtered_tokens)

    if category == "title":
        get_post_entity(id, lemmatized_sentence)

    return lemmatized_sentence<jupyter_output><empty_output><jupyter_text>Normalizing the post invokes `get_post_entity`, triggering the extraction of named entities into csv.  Using PRAW to connect to Reddit[PRAW](https://praw.readthedocs.io/en/latest/) is the Python Reddit API Wrapper that we used to download the submissions from the _subreddits_.I'm not going into the details of the data scraping.  **You can check out how you can get the API key from my article on [Reddit Scraping](https://medium.com/atoti/reddit-data-analytics-trilogy-1-data-scraping-with-praw-89f3df7dfa90eb02)**.  While the below variables are placeholders for the Reddit API keys, it is __NOT recommended__ to add these variables in your source files and commit them. Instead, secure your API keys for example, using environment variable instead. Refer to the article on [securely storing configuration credentials](http://veekaybee.github.io/2020/02/25/secrets/).<jupyter_code># reddit_cid = "placeholder - cid"
# reddit_csec = "placeholder - csec"
# reddit_ua = "placeholder - ua"

reddit_cid = "vcJSENEMmnCztA"
reddit_csec = "CIxOTX27NWFg0m2_kjrZvqHHiWk"
reddit_ua = "atoti-scrape"
# create a reddit connection
reddit = praw.Reddit(
    client_id=reddit_cid, client_secret=reddit_csec, user_agent=reddit_ua
)<jupyter_output><empty_output><jupyter_text>In order to do more analysis, I created a function to get the latest postings. This function will allow me to configure the number of postings to download, and from which subreddit to download the postings from.Unlike the original article, I decided to format `created` into _datetime_ and splitted it into `created_date` and `created_time` as I figured this would allow me to have consolidated view on date level or drill down to the time level.Referencing the [PRAW documentation](https://praw.readthedocs.io/en/latest/code_overview/models/submission.htmlsubmission) for the submission attributes, I added _id_ and _author_ to the list as well. Important to note is that for real-time, I will be polling the latest 100 postings at a 30 seconds intervals. Potentially there will be duplicated postings across the pollings. Storing the _id_ allows me to **leverage on atoti's capability to update existing posts instead of inserting duplicated postings**.<jupyter_code>def get_latest_posts(subreddit, count):
    # list for df conversion
    _posts = []
    # return 100 new posts from wallstreetbets
    new_bets = reddit.subreddit(subreddit).new(limit=count)
    # return the important attributes
    for post in new_bets:
        # normalize the post and perform NLP named entities extraction
        lemmatized_post = normalize("post", post.id, post.selftext)
        lemmatized_title = normalize("title", post.id, post.title)
        _posts.append(
            [
                post.id,
                post.author,
                post.title,
                post.score,
                post.num_comments,
                post.selftext,
                lemmatized_post,
                lemmatized_title,
                post.created,
                post.pinned,
                post.total_awards_received,
            ]
        )

    # create a dataframe
    _posts = pd.DataFrame(
        _posts,
        columns=[
            "id",
            "author",
            "title",
            "score",
            "comments",
            "post",
            "lemmatized post",
            "lemmatized title",
            "created",
            "pinned",
            "total awards",
        ],
    )

    _posts["created"] = pd.to_datetime(_posts["created"], unit="s")
    _posts["created date"] = pd.to_datetime(_posts["created"], unit="s").dt.date
    _posts["created time"] = pd.to_datetime(_posts["created"], unit="s").dt.time
    _posts["subreddit"] = subreddit

    return _posts<jupyter_output><empty_output><jupyter_text>To kickstart, I test downloaded the newest 500 post from the subreddit _wallstreetbets_.  This data will be loaded into atoti's **in-memory datacube** later on. This means that the **amount of data that can be loaded is limited only by the hosting machine's capacity**. You could adjust the number of postings you want to look at according to your machine's memory capacity.<jupyter_code>posts = get_latest_posts("wallstreetbets", 500)
entity_file.flush()

# check the number of posts retrieved
len(posts)<jupyter_output><empty_output><jupyter_text>I adapted the below code from the original article and used the Reddit's title instead of post for NLP. Mainly because the Subreddits that I chose has more text in the title than the post.  The function uses Pandas dataframe to compute some simple statistics about the text in the posts.<jupyter_code>def get_nlp_features(_posts):
    # copy the dataframe
    df = _posts.copy()
    # count words in post
    df["words"] = df["title"].apply(lambda x: len(x.split()))
    # count characters in post
    df["chars"] = df["title"].apply(lambda x: len(x.replace(" ", "")))
    # calculate word density
    df["word density"] = (df["words"] / (df["chars"] + 1)).round(3)
    # count unique words
    df["unique words"] = df["title"].apply(lambda x: len(set(w for w in x.split())))
    # percent of unique words
    df["unique density"] = (df["unique words"] / df["words"]).round(3)

    return df<jupyter_output><empty_output><jupyter_text>By running the function above on the posts that I have downloaded, we will get a dataset with similar structure to the original article.<jupyter_code>df = get_nlp_features(posts)
df.head(3)<jupyter_output><empty_output><jupyter_text>atoti cube creation for NLP analysisFinally we can get started with atoti. Let's create a session with the [sampling_mode](https://docs.atoti.io/0.4.2/lib/atoti.htmlmodule-atoti.sampling) to set full, so that I can see all the data once it is loaded.<jupyter_code>config = create_config(metadata_db="./metadata.db", sampling_mode=tt.sampling.FULL)
session = tt.create_session(config=config)<jupyter_output><empty_output><jupyter_text>atoti datastoresI will create a `Reddit Posts` store for the data as per the original article.<jupyter_code>reddit_store = session.read_pandas(df, keys=["id"], store_name="Reddit_Posts")
reddit_store.head(3)<jupyter_output><empty_output><jupyter_text>For my experiment with spaCy, I create a `Entities` store to store the named entities.  Below, I set `watch=True` for the entities.csv such that any update to the file will trigger an upload to the datastore.<jupyter_code>entity_store = session.read_csv(
    entity_file_name,
    keys=["id", "category", "text"],
    store_name="Entities",
    watch=True,
    sep="|",
)
entity_store.head()<jupyter_output><empty_output><jupyter_text>I join the `Entities` store to `Reddit Posts` which I decided to use as the base store.  Note that atoti automatically joins columns with the same name across both stores. You could define your own [mapping](https://docs.atoti.io/0.4.2/lib/atoti.html?highlight=joinatoti.store.Store.join) otherwise.<jupyter_code>reddit_store.join(entity_store)<jupyter_output><empty_output><jupyter_text>atoti cube creation - No measuresBy default, atoti will automatically create hierarchies and measures from the datastore based on the data type of the columns.  In this case, I prefer to manually create my measures, hence I set the [mode](https://docs.atoti.io/0.4.2/lib/atoti.html?highlight=create_cubeatoti.session.Session.create_cube) of creation to `no_measures`<jupyter_code>cube = session.create_cube(reddit_store, "Reddit", mode="no_measures")
cube.schema<jupyter_output><empty_output><jupyter_text>Take a look at the structure of the cube created. Only a `Count` measure is created.<jupyter_code>m = cube.measures
l = cube.levels
h = cube.hierarchies
cube<jupyter_output><empty_output><jupyter_text>Managing hierarchiesSince I am going to create my own measures, I will delete the numeric columns from my hierarchies to avoid confusion with the measures that I will be creating.<jupyter_code>del h["text count"]
del h["score"]
del h["comments"]
del h["total awards"]
del h["words"]
del h["chars"]
del h["unique words"]<jupyter_output><empty_output><jupyter_text>Measures creationIn default `auto` mode, atoti will create a `MEAN` and a `SUM` measure.  However, as I only require `sum` in my simple use case and in order to demonstrate how simple aggregation can be done with atoti, I created the below measures instead.<jupyter_code>m["Text Count"] = tt.agg.sum(entity_store["text count"])
m["Score"] = tt.agg.sum(reddit_store["score"])
m["Comments"] = tt.agg.sum(reddit_store["comments"])
m["Total awards"] = tt.agg.sum(reddit_store["total awards"])
m["Words"] = tt.agg.sum(reddit_store["words"])
m["Chars"] = tt.agg.sum(reddit_store["chars"])
m["Unique words"] = tt.agg.sum(reddit_store["unique words"])
m["Word density"] = tt.agg.sum(reddit_store["word density"])
m["Unique density"] = tt.agg.sum(reddit_store["unique density"])<jupyter_output><empty_output><jupyter_text>Distinct measureI use the [`count_distinct`](https://docs.atoti.io/0.4.2/lib/atoti.html?highlight=count_distinctatoti.agg.count_distinct) function to compute the number of days the data span across.<jupyter_code>m["Number days"] = tt.agg.count_distinct(reddit_store["created date"])<jupyter_output><empty_output><jupyter_text>Cumulative measureAggregation with the [`cumulative`](https://docs.atoti.io/0.4.2/lib/atoti.scope.html?highlight=cumulativeatoti.scope.cumulative) scope allows me to see the trend of the posting over the `created` level. By setting the parameter `dense=True`, we can see a continuous plot of the trend even if there are no postings on days in between.<jupyter_code>m["Cumulative Count"] = tt.agg.sum(
    m["contributors.COUNT"], scope=tt.scope.cumulative(l["created"], dense=True)
)
session.visualize("Posting Trends")<jupyter_output><empty_output><jupyter_text>atoti Pivot tableatoti is currently going through UX and performance upgrade. Hang on with us as more features for the Pivot table will come with future release.  atoti Pivot table against Dash DataTableatoti is currently going through UX and performance upgrade. Hang on with us as more features for the Pivot table will come with future release.  Let's see how atoti data table fare compared to the [Dash DataTable](https://medium.com/swlh/dashboards-in-python-using-dash-creating-a-data-table-using-data-from-reddit-1d6c0cecb4bdbe7c) in the original article before the upgrade happened? _Maybe_ Dash Datatable has evolved since the article was last published, let's just enjoy the atoti table.<jupyter_code>session.visualize()<jupyter_output><empty_output><jupyter_text>I can easily adjust the height of the table or the width of the columns. With the atoti Editor, I can easily:- add or remove the levels and measures- filter the data- sort the measures- format the columnsRefer to the GIF below to see how these could be done. Creating table Table adjustment Cell formatting and miscI can toggle the table from a tabular table to pivot table. Pivot table allows me to drill down on any hierarchies. The measures are computed on the fly as the level changes.<jupyter_code>session.visualize()<jupyter_output><empty_output><jupyter_text>Real-time dashboarding with atotiAt the moment, I have only loaded data from the subreddit _wallstreetbets_. In the next few cell, I have created functions that allow me to poll the latest 200 postings from some other subreddit groups:- news- worldnews- politics- technology- wallstreetbetsLet's switch over to a dashboard that I have prepared in advance by accessing the url from the next cell. After loading the dashboard, trigger the rest of the cells to start the data polling. You can then go back to the dashboard to see it getting refreshed each time new data is downloaded.<jupyter_code>session.url + "/#/dashboard/c7c"<jupyter_output><empty_output><jupyter_text>I used [`store.load_pandas`](https://docs.atoti.io/0.4.2/lib/atoti.html?highlight=load_pandasatoti.store.Store.load_pandas) to load the Pandas dataframe incrementally into the `Reddit Posts` store. For the `Entities` store, the csv file is on `watch` mode, hence any update will trigger data to be uploaded into the datastore. Based on the stores' keys, existing data will be updated and new data will be inserted.<jupyter_code>def load_new_subreddit(subreddit):
    posts_news = get_latest_posts(subreddit, 200)
    df_news = get_nlp_features(posts_news)

    reddit_store.load_pandas(df_news)<jupyter_output><empty_output><jupyter_text>For demo purpose, I have only set the polling to iterate 10 times with intervals of 30s.<jupyter_code>import queue
import threading

should_publish = threading.Event()
update_queue = queue.Queue()


def start_publisher():
    # Do 10 iterations with 30s interval
    starttime = time.time()
    print("Start polling", starttime)
    poll_iteration = 1

    while poll_iteration <= 10:
        print("\rpublishing update ", end="")
        update_queue.put((poll_iteration))
        poll_iteration += 1
        time.sleep(30)
        print("\rawaiting for publishing update", end="")
        should_publish.wait()
        update_queue.join()


def start_update_listener():
    while True:
        poll_iteration = update_queue.get()
        print(poll_iteration, ">>> Start polling from subreddit... ")
        # poll from subreddit
        load_new_subreddit("news")
        print(poll_iteration, "........ Loaded news ")

        load_new_subreddit("worldnews")
        print(poll_iteration, "........ Loaded worldnews ")

        load_new_subreddit("politics")
        print(poll_iteration, "........ Loaded politics ")

        load_new_subreddit("technology")
        print(poll_iteration, "........ Loaded technology ")

        load_new_subreddit("wallstreetbets")
        print(poll_iteration, "........ Loaded wallstreetbets ")
        entity_file.flush()
        print(poll_iteration, ">>> End polling from subreddit... ")
        update_queue.task_done()


listener_thread = threading.Thread(target=start_update_listener, daemon=True)
publisher_thread = threading.Thread(target=start_publisher, daemon=True)
publisher_thread.start()
listener_thread.start()
# start publishing
should_publish.set()<jupyter_output>Start polling 1616740512.413595
publishing update 1 >>> Start polling from subreddit...<jupyter_text>If you would like to stop the data polling before the 10 iterations end, run the below cell.<jupyter_code># pause publishing
should_publish.clear()
entity_file.close()<jupyter_output>Exception in thread Thread-8:
Traceback (most recent call last):
  File "C:\Apps\miniconda3\envs\atoti\lib\threading.py", line 954, in _bootstrap_inner
    self.run()
  File "C:\Apps\miniconda3\envs\atoti\lib\threading.py", line 892, in run
    self._target(*self._args, **self._kwargs)
  File "<ipython-input-31-cf3ea888647e>", line 29, in start_update_listener
  File "<ipython-input-30-d195cc565785>", line 2, in load_new_subreddit
  File "<ipython-input-10-f5e80723cc2c>", line 10, in get_latest_posts
  File "<ipython-input-7-e5afd54dffe5>", line 23, in normalize
  File "<ipython-input-6-84034344cb64>", line 67, in get_post_entity
ValueError: I/O operation on closed file.
{'fraction_non_alphanumeric': 0.09804809564784203, 'fraction_numerical': 0.008151804718989176, 'mean_word_length': 4.172639962520496, 'pattern_counts': {'":': 7, '<': 109, '<?xml version=': 0, '>': 118, 'https://': 26, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '26', 'dir': 'jupyter-structured-clean-dedup', 'id': '303213', 'license': 'Apache-2.0', 'n_tokens_mistral': 6827, 'n_tokens_neox': 6547, 'n_words': 2138, 'path': 'notebooks/reddit/main.ipynb', 'repo_name': 'tiphanieqch/notebooks', 'size': '22081'}