Amazon Athena: Beyond The Basics – Part 1

def get_parser():
“””Get parser for command line arguments.”””
parser = argparse.ArgumentParser(description=”Twitter Downloader”)
parser.add_argument(“-q”,
“–query”,
dest=”query”,
help=”Query/Filter”,
default=’-‘)
parser.add_argument(“-d”,
“–data-dir”,
dest=”data_dir”,
help=”Output/Data Directory”)
return parser

class MyListener(StreamListener):
“””Custom StreamListener for streaming data.”””

def __init__(self, data_dir, query):
query_fname = format_filename(query)
self.outfile = “%s/stream_%s.json” % (data_dir, query_fname)

def on_data(self, data):
try:
result = send_record_to_firehose(data)
print(result)
except BaseException as e:
print(“Error on_data: %s” % str(e))
time.sleep(5)
return True

def on_error(self, status):
print(“Error with status:” + str(status))
if status == 420:
print(“You are being rate limited!!!.”)
return True

def send_record_to_firehose(data):
“”” Sends Json response from tweeter to Kinesis Firehose Delivery Stream
Arguments:
data — json file from tweeter
Return:
String — json response from Kinesis Firehos
“””
client = boto3.client(‘firehose’)
response = client.put_record(
DeliveryStreamName=’twitter-to-s3′,
Record={
‘Data’: data
}
)
return response

def format_filename(fname):
“””Convert file name into a safe string.

Arguments:
fname — the file name to convert
Return:
String — converted file name
“””
return ”.join(convert_valid(one_char) for one_char in fname)

def convert_valid(one_char):
“””Convert a character into ‘_’ if invalid.

Arguments:
one_char — the char to convert
Return:
Character — converted char
“””
valid_chars = “-_.%s%s” % (string.ascii_letters, string.digits)
if one_char in valid_chars:
return one_char
else:
return ‘_’

@classmethod
def parse(cls, api, raw):
status = cls.first_parse(api, raw)
setattr(status, ‘json’, json.dumps(raw))
return status

if __name__ == ‘__main__’:
parser = get_parser()
args = parser.parse_args()
auth = OAuthHandler(config.consumer_key, config.consumer_secret)
auth.set_access_token(config.access_token, config.access_secret)
api = tweepy.API(auth)
# Added this logic to reconnect if it fails
while True:
try:
twitter_stream = Stream(auth, MyListener(args.data_dir, args.query))
twitter_stream.filter(track=[args.query])
except Exception:
# Sure??, let’s reconnect and keep tracking
continue

CREATE EXTERNAL TABLE IF NOT EXISTS tweets (
created_at string ,
id string ,
id_str string ,
text string ,
display_text_range ARRAY,
source string ,
truncated string ,
user struct< id:string , id_str:string , name:string , screen_name:string , location:string , description:string , protected:string , verified:string , followers_count:string , friends_count:string , listed_count:string , favourites_count:string , statuses_count:string , created_at:string , utc_offset:string , time_zone:string , geo_enabled:string , lang:string>,
is_quote_status string,
extended_tweet STRUCT< full_text:string, display_text_range:ARRAY,
entities:STRUCT< media:ARRAY<STRUCT< id:string, id_str:string, indices:ARRAY,
media_url:string,
media_url_https:string,
url:string,
display_url:string,
expanded_url:string,
type:string,
sizes:STRUCT< small:STRUCT<w:string, h:string, resize:string>,
thumb:STRUCT<w:string, h:string, resize:string>>>>>>,
retweet_count string,
favorite_count string,
retweeted_status STRUCT< retweet_count:string, text:string>,
entities STRUCT< urls:ARRAY<STRUCT<url:string, expanded_url:string, display_url:string, indices:ARRAY>>,
user_mentions:ARRAY<STRUCT>,
hashtags:ARRAY>,
favorited string,
retweeted string,
possibly_sensitive string,
filter_level string,
lang string)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://twitter-to-s3/json/';

Detailed documentation is available from the following link to establish a connection to Athena from a client tool such as SQL Workbench:

http://docs.aws.amazon.com/athena/latest/ug/connect-with-jdbc.html

SELECT ht.text,count(*)
FROM tweets
CROSS JOIN UNNEST (entities.hashtags) AS t(ht)
GROUP BY ht.text
HAVING count(*)>100
ORDER by count(*) desc

SELECT user.screen_name,user.name,max(user.followers_count),count(*)
FROM tweets
WHERE user.verified='true'
GROUP BY user.screen_name,user.name
ORDER BY cast(max(user.followers_count) as integer) DESC