Building a Twitter Data Pipeline with Apache Airflow|Mini project for Data Engineering
In this video, we'll explore how to set up a robust data pipeline using Apache Airflow to extract real-time data from Twitter, store it in Amazon S3, and finally load it into Amazon Redshift for analysis. Twitter is a treasure trove of valuable insights, and with the power of Airflow, we can automate the entire process and ensure data consistency and reliability. Join us as we walk through the step-by-step implementation of this Twitter data pipeline, covering the setup of Airflow, configuring Twitter API credentials, setting up S3 buckets, and leveraging Redshift's data warehousing capabilities. By the end of this tutorial, you'll have a fully functional data pipeline ready to unlock the potential of Twitter data in your analytics project
=======
#code
---------
create table in redshift:-
create table tweet(
user varchar(max),
text varchar(max),
favorite_count varchar(max),
retweet_count varchar(max),
created_at varchar(max)
)
----------------------
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.operators.python_operator import PythonOperator
import tweepy
import pandas as pd
import boto3
import psycopg2
def copy_s3_to_redshift():
conn_string = "dbname='{}' port='{}' user='{}' password='{}' host='{}'"\
.format('dev','5439','admin', 'Sumit123456789', 'default.655121573728.us-east-1.redshift-serverless.amazonaws.com')
print(conn_string)
con = psycopg2.connect(conn_string)
cur = con.cursor()
sqlQuery='''copy public.tweet
from 's3://ec3testbucket007/refined_tweets_18_05.csv'
iam_role 'arn:aws:iam::655121573728:role/service-role/AmazonRedshift-CommandsAccessRole-20230323T225145'
delimiter ','
IGNOREHEADER as 1
csv;'''
#sqlQuery="insert into test.dept values('d1221',100)"
print(sqlQuery)
cur.execute(sqlQuery)
def run_twitter_etl():
access_token = "3251395693-2Uw9KG2W7BXXSqZFzW0I7zB6oO4lSdylqGF2r6M"
access_token_secret = "EzyEwcnWL1LMVIrsS7S9qP9yuRuaBKmnrERMOtKnDQtDU"
consumer_key = "ELEswKF533rwKpouOxf958pSP"
consumer_secret = "LPNKThG9MREllSWKkJvHQlFn6exObt5la68RD7p13fBr1lBZt3"
auth = tweepy.OAuth1UserHandler(
consumer_key, consumer_secret, access_token, access_token_secret
)
api = tweepy.API(auth)
tweets = api.user_timeline(screen_name='@elonmusk',count=200,include_rts = False,tweet_mode = 'extended')
list = []
for tweet in tweets:
text = tweet._json["full_text"]
refined_tweet = {"user": tweet.user.screen_name,
'text' : text,
'favorite_count' : tweet.favorite_count,
'retweet_count' : tweet.retweet_count,
'created_at' : tweet.created_at}
list.append(refined_tweet)
df = pd.DataFrame(list)
print("print dataframe")
print(df.head(10))
df.to_csv('/opt/airflow/dags/refined_tweets.csv',index=False)
def upload_into_s3(file_name, s3_bucket, s3_key):
s3 = boto3.client("s3",aws_access_key_id="AKIAZRCCJ45QGOCJ67UX",aws_secret_access_key="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")
s3.upload_file(
file_name,
s3_bucket,
s3_key
)
with DAG(dag_id="twitter_python_api_v5",default_args={"owner": "sumit"},start_date=datetime(2023, 5, 18), schedule="0 0 * * *",tags=["python_test", ]) as dag:
download_data_from_twitter = PythonOperator(
task_id='download_data_from_twitter',
python_callable=run_twitter_etl
)
upload_data_into_s3 = PythonOperator(
task_id='upload_data_into_s3',
python_callable=upload_into_s3,
op_args=['/opt/airflow/dags/refined_tweets.csv', 'ec3testbucket007', 'refined_tweets_18_05.csv']
)
copy_s3_to_redshift = PythonOperator(
task_id='copy_s3_to_redshift',
python_callable=copy_s3_to_redshift
)
download_data_from_twitter (double gt) upload_data_into_s3 (double gt)copy_s3_to_redshift