Reworking Python for loop to ETL Workflow - Using Luigi, Airflow, etc

Viewed 162

I'm currently experimenting with different python workflow techniques and I have a nested for loop that I want to convert into an automated workflow. I've been trying to use luigi, but I am unable to figure out a successful workflow that takes in two datasets that depend on each other and outputs both dataset chunks CSV. Every luigi example that I've seen so far takes in data in one step, aggregates, then writes the output.

In my example, I want to bring in a daily NBA scoreboard from the API, store that in a CSV; then using that scoreboard data to bring in the stats (or boxscore) of each of the games for that day and store those in separate CSVs.

The basic nested for loop that achieves what I want is the following:

import re
import requests
import pandas as pd
from pandas.io.json import json_normalize
import os

if not os.path.exists('data/'):
    os.makedirs('data/')
if not os.path.exists('data/games/'):
    os.makedirs('data/games/')
if not os.path.exists('data/boxscores/'):
    os.makedirs('data/boxscores/')

dates = ['2020-03-01', '2020-03-02'] # and so on...
for date in dates:
    print('*** DATE: {}'.format(date))
    date = re.sub( '-', '', date)
    print('*** MODIFIED DATE: {}'.format(date))
    response = requests.get('http://data.nba.net/json/cms/noseason/scoreboard/{}/games.json'.format(date))
    df = pd.read_json(response.text)
    games_df = json_normalize(df[df.index == 'games']['sports_content'][0]['game'])
    games_df.to_csv('data/games/games_{}.csv'.format(date), index=False)
    for id in games_df.id.tolist():
        print('*** GAME ID: {}'.format(id))
        response = requests.get("http://data.nba.net/10s/prod/v1/{0}/{1}_boxscore.json".format(date, id))
        df = pd.read_json(response.text)
        df = json_normalize(df[df.index == 'activePlayers']['stats'][0])
        boxscore = df[['personId', 'firstName', 'lastName', 'teamId',
               'min', 'points', 'fgm', 'fga',
               'ftm', 'fta', 'tpm', 'tpa', 'offReb', 'defReb',
               'totReb', 'assists', 'pFouls', 'steals', 'turnovers', 'blocks', 'plusMinus']]
        boxscore['min'] = boxscore['min'].str.split(":", expand = True)[0]
        boxscore.to_csv('data/boxscores/boxscore_{}.csv'.format(id), index=False) 

I want to avoid using for loops and I like how luigi's framework will avoid re-building days that you already have built. Does anyone have any suggestions or links on how to build a pipeline like this? I'm open to switching from luigi to Airflow if it is more intuitive.

0 Answers
Related