dagster_etl_pipeline_hacker_news_word_frequency_analysis.py

1import requests
2import pandas as pd
3from dagster import asset, Definitions
4
5@asset
6def topstory_ids():
7    newstories_url = "https://hacker-news.firebaseio.com/v0/topstories.json"
8    top_new_story_ids = requests.get(newstories_url).json()[:100]
9    return top_new_story_ids
10
11@asset
12def topstories(topstory_ids):
13    results = []
14    for item_id in topstory_ids:
15        item = requests.get(f"https://hacker-news.firebaseio.com/v0/item/{item_id}.json").json()
16        results.append(item)
17
18    df = pd.DataFrame(results)
19    return df
20
21@asset
22def most_frequent_words(topstories):
23    # This asset depends on the 'topstories' asset
24    stopwords = ["a", "the", "an", "of", "and", "in", "to", "for", "is", "on", "that", "by", "with", "as"]
25    
26    # Simple word frequency calculation
27    words = " ".join(topstories["title"].dropna()).lower().split()
28    word_counts = {}
29    for word in words:
30        if word not in stopwords and len(word) > 2:
31            word_counts[word] = word_counts.get(word, 0) + 1
32            
33    # Return top 10 most frequent words
34    return sorted(word_counts.items(), key=lambda x: x[1], reverse=True)[:10]
35
36# Definitions object tells Dagster which assets to load
37defs = Definitions(
38    assets=[topstory_ids, topstories, most_frequent_words],
39)