Initial commit

data/*.json.gz

*.png

config/*

*/config/*

*.db

*.log

*~

*.pyc

venv/*

*.*/

*.pid

*.db

real_variables.yml

data/tweets/*.csv

# airflow configs / logs

airflow.cfg

unittests.cfg

*/logs/*

django/*

## Data Pipelines with Python (video edition)

Welcome to the code repository for [Data Pipelines with Python]()! If you have any questions reach out to @kjam on Twitter or GitHub.

### Code Structure

Most of the code covered in the videos is here; but not all of it. I highly recommend you take time to type out all the code along with the videos and simply use these scripts to "double check" or remind yourself of the work you've already completed.

### Installation

If you are using Python2, use the requirements.txt file. If you are using Python3, use the py3_requirements.txt file. 

```pip install -r requirements.txt```

or 

```pip install -r py3_requirements.txt```

### Yahoo Finance API

There is a [good writeup in German for the Finance API](http://brusdeylins.info/tips_and_tricks/yahoo-finance-api/) which I used as a starting point to download newer-time data.

### Python2 v. Python3

This repository is primarily compliant for both versions. Please let me know if you run into any bugs!

### Ansible Playbook

To use as a template rather than as a direct template, I've included a working playbook in the deploy folder. If you try and run it directly, you will likely receive some errors. Please read through the notebook and take a look at the directives and determine which you need and which you don't. It also requires a .ssh/authorized_hosts file as well as a config file located in `celeryapp/config/prod.cfg`. If you run into other errors, I highly recommend reading through [the Ansible

documentation](http://docs.ansible.com/ansible/) or searching on StackOverflow.

### Corrections?

If you find any issues in these code examples, feel free to submit an Issue or Pull Request. I appreciate your input!

### Questions?

Reach out to @kjam on Twitter or GitHub. @kjam is also often on freenode. :)

""" Simple example of creating subdags and generating work dynamically"""

from airflow import DAG

from airflow.hooks import SqliteHook

from airflow.operators import BashOperator, EmailOperator, SubDagOperator, \

    PythonOperator, BranchPythonOperator

from twitter_airflow import search_twitter, RAW_TWEET_DIR

from subdags.twitter_subdag import subdag

from datetime import datetime, timedelta

import pandas as pd

import re

import random

SEARCH_TERMS = ['#python', '#pydata', '#airflow', 'data wrangling',

                'data pipelines']

default_args = {

    'owner': 'admin',

    'depends_on_past': False,

    'start_date': datetime.now() - timedelta(days=4),

    'retries': 1,

    'retry_delay': timedelta(minutes=5),

}

dag = DAG('generate_twitter_dags', default_args=default_args,

          schedule_interval='@daily')

def fill_terms(my_terms=SEARCH_TERMS, **kwargs):

    """ Fill sqlite database with a few search terms. """

    sqlite = SqliteHook('twitter_sqlite')

    conn = sqlite.get_conn()

    df = pd.DataFrame(my_terms, columns=['search_term'])

    try:

        df.to_sql('twitter_terms', conn)

    except ValueError:

        # table already exists

        pass

def generate_search_terms(**kwargs):

    """ Generate subdag to search twitter for terms. """

    sqlite = SqliteHook('twitter_sqlite')

    conn = sqlite.get_conn()

    query = "select * from twitter_terms"

    df = pd.read_sql_query(query, conn)

    return random.choice([

        'search_{}_twitter'.format(re.sub(r'\W+', '', t))

        for t in df.search_term.values])

fill_search_terms = PythonOperator(task_id='fill_terms',

                                   provide_context=True,

                                   python_callable=fill_terms,

                                   dag=dag)

gen_search_terms = BranchPythonOperator(task_id='generate_search_terms',

                                        provide_context=True,

                                        python_callable=generate_search_terms,

                                        dag=dag)

email_links = EmailOperator(task_id='email_best_links',

                            to='MYEMAIL@MYSITE.com',

                            subject='Latest popular links',

                            html_content='Check out the latest!!',

                            files=['{}/latest_links.txt'.format(RAW_TWEET_DIR)],

                            dag=dag)

sub = SubDagOperator(subdag=subdag,

                     task_id='insert_and_id_pop',

                     trigger_rule='one_success',

                     dag=dag)

clear_latest = BashOperator(bash_command='rm -rf {}/latest_links.txt'.format(

    RAW_TWEET_DIR), task_id='clear_latest', dag=dag)

gen_search_terms.set_upstream(fill_search_terms)

for term in SEARCH_TERMS:

    term_without_punctuation = re.sub(r'\W+', '', term)

    simple_search = PythonOperator(

        task_id='search_{}_twitter'.format(term_without_punctuation),

        provide_context=True,

        python_callable=search_twitter,

        dag=dag,

        params={'query': term})

    simple_search.set_upstream(gen_search_terms)

    simple_search.set_downstream(sub)

sub.set_downstream(email_links)

email_links.set_downstream(clear_latest)

""" Simple subdag example """

from airflow import DAG

from airflow.operators import PythonOperator

from twitter_airflow import csv_to_sqlite, identify_popular_links

from datetime import datetime, timedelta

default_args = {

    'owner': 'admin',

    'depends_on_past': False,

    'start_date': datetime(2016, 1, 1),

    'retries': 1,

    'retry_delay': timedelta(minutes=5),

}

subdag = DAG('generate_twitter_dags.insert_and_id_pop',

             default_args=default_args)

move_tweets_to_sqlite = PythonOperator(task_id='csv_to_sqlite',

                                       provide_context=True,

                                       python_callable=csv_to_sqlite,

                                       dag=subdag)

id_popular = PythonOperator(task_id='identify_popular_links',

                            provide_context=True,

                            python_callable=identify_popular_links,

                            dag=subdag,

                            params={'write_mode': 'a'})

id_popular.set_upstream(move_tweets_to_sqlite)

""" Simple Airflow data pipeline example using Twitter API """

from airflow import DAG

from airflow.operators import EmailOperator, PythonOperator

from airflow.hooks import SqliteHook

from tweepy import API, Cursor, OAuthHandler

from configparser import ConfigParser

from csv import DictWriter, writer

from collections import Counter

from datetime import datetime, timedelta

import ast

import itertools

import glob

import shutil

import pandas as pd

import os.path

RAW_TWEET_DIR = os.path.abspath(os.path.join(__file__, '../../../data/tweets/'))

CONFIG_FILE = os.path.abspath(os.path.join(__file__,

                                           '../../../config/prod.cfg'))

MAX_TWEEPY_PAGE = 300

default_args = {

    'owner': 'admin',

    'depends_on_past': False,

    'start_date': datetime.now() - timedelta(days=4),

    'retries': 1,

    'retry_delay': timedelta(minutes=5),

}

dag = DAG('twitter_links', default_args=default_args,

          schedule_interval='@daily')

def extract_tweet_data(tweepy_obj, query):

    """ Extract relevant and serializable data from a tweepy Tweet object

        params:

            tweepy_obj: Tweepy Tweet Object

            query: str

        returns dict

    """

    return {

        'user_id': tweepy_obj.user.id,

        'user_name': tweepy_obj.user.name,

        'user_screenname': tweepy_obj.user.screen_name,

        'user_url': tweepy_obj.user.url,

        'user_description': tweepy_obj.user.description,

        'user_followers': tweepy_obj.user.followers_count,

        'user_friends': tweepy_obj.user.friends_count,

        'created': tweepy_obj.created_at.isoformat(),

        'text': tweepy_obj.text,

        'hashtags': [ht.get('text') for ht in

                     tweepy_obj.entities.get('hashtags')],

        'mentions': [(um.get('id'), um.get('screen_name')) for um in

                     tweepy_obj.entities.get('user_mentions')],

        'urls': [url.get('expanded_url') for url in

                 tweepy_obj.entities.get('urls')],

        'tweet_id': tweepy_obj.id,

        'is_quote_status': tweepy_obj.is_quote_status,

        'favorite_count': tweepy_obj.favorite_count,

        'retweet_count': tweepy_obj.retweet_count,

        'reply_status_id': tweepy_obj.in_reply_to_status_id,

        'lang': tweepy_obj.lang,

        'source': tweepy_obj.source,

        'location': tweepy_obj.coordinates,

        'query': query,

    }

def search_twitter(**kwargs):

    """ simple search for a query in public tweets"""

    query = kwargs.get('params').get('query')

    config = ConfigParser()

    config.read(CONFIG_FILE)

    auth = OAuthHandler(config.get('twitter', 'consumer_key'),

                        config.get('twitter', 'consumer_secret'))

    auth.set_access_token(config.get('twitter', 'access_token'),

                          config.get('twitter', 'access_token_secret'))

    api = API(auth)

    all_tweets = []

    page_num = 0

    since_date = datetime.strptime(

        kwargs.get('ds'), '%Y-%m-%d').date() - timedelta(days=1)

    query += ' since:{} until:{}'.format(since_date.strftime('%Y-%m-%d'),

                                         kwargs.get('ds'))

    print('searching twitter with: %s' % query)

    for page in Cursor(api.search, q=query, monitor_rate_limit=True,

                       wait_on_rate_limit=True).pages():

        all_tweets.extend([extract_tweet_data(t, query) for t in page])

        page_num += 1

        if page_num > MAX_TWEEPY_PAGE:

            break

    # if it's an empty list, stop here

    if not len(all_tweets):

        return

    filename = '{}/{}_{}.csv'.format(

        RAW_TWEET_DIR, query, datetime.now().strftime('%m%d%Y%H%M%S'))

    with open(filename, 'w') as raw_file:

        raw_wrtr = DictWriter(raw_file, fieldnames=all_tweets[0].keys())

        raw_wrtr.writeheader()

        raw_wrtr.writerows(all_tweets)

def csv_to_sqlite(directory=RAW_TWEET_DIR, **kwargs):

    """ Very basic csv to sqlite pipeline using pandas

        params:

            directory: str (file path to csv files)

    """

    sqlite = SqliteHook('twitter_sqlite')

    conn = sqlite.get_conn()

    for fname in glob.glob('{}/*.csv'.format(directory)):

        if '_read' not in fname:

            try:

                df = pd.read_csv(fname)

                df.to_sql('tweets', conn, if_exists='append', index=False)

                shutil.move(fname, fname.replace('.csv', '_read.csv'))

            except pd.io.common.EmptyDataError:

                # probably an io error with another task / open file

                continue

def identify_popular_links(directory=RAW_TWEET_DIR, write_mode='w', **kwargs):

    """ Identify the most popular links from the last day of tweest in the db

        Writes them to latest_links.txt in the RAW_TWEET_DIR

        (or directory kwarg)

    """

    sqlite = SqliteHook('twitter_sqlite')

    conn = sqlite.get_conn()

    query = """select * from tweets where

    created > date('now', '-1 days') and urls is not null

    order by favorite_count"""

    df = pd.read_sql_query(query, conn)

    df.urls = df.urls.map(ast.literal_eval)

    cntr = Counter(itertools.chain.from_iterable(df.urls.values))

    with open('{}/latest_links.txt'.format(directory), write_mode) as latest:

        wrtr = writer(latest)

        wrtr.writerow(['url', 'count'])

        wrtr.writerows(cntr.most_common(5))

simple_search = PythonOperator(task_id='search_twitter',

                               provide_context=True,

                               python_callable=search_twitter,

                               dag=dag,

                               params={'query': '#python'})

move_tweets_to_sqlite = PythonOperator(task_id='csv_to_sqlite',

                                       provide_context=True,

                                       python_callable=csv_to_sqlite,

                                       dag=dag)

id_popular = PythonOperator(task_id='identify_popular_links',

                            provide_context=True,

                            python_callable=identify_popular_links,

                            dag=dag)

email_links = EmailOperator(task_id='email_best_links',

                            to='katharine@kjamistan.com',

                            subject='Latest popular links',

                            html_content='Check out the latest!!',

                            files=['{}/latest_links.txt'.format(RAW_TWEET_DIR)],

                            dag=dag)

simple_search.set_downstream(move_tweets_to_sqlite)

id_popular.set_upstream(move_tweets_to_sqlite)

email_links.set_upstream(id_popular)