xYv6KiDq

1#!/usr/bin/python
2
3"""
4ETL for Fastly views.
5"""
6
7import datetime
8import re
9
10import luigi
11
12import athena
13
14
15s3_table_data_path = "s3://fastly-logs/date=%s/"
16
17
18class UpdateFastlyPartitions(athena.AthenaQuery):
19    """
20    Repair Fastly partitions.
21    """
22
23    table_name = "fastly"
24    results_path_template = "repair-table-fastly/%s"
25
26    def run(self):
27        # An alternative approach would be to simply use an
28        # "msck repair table fastly" statement but this is very slow at Athena.
29
30        client = luigi.s3.S3Client()
31
32        paths = {}
33        for date in [self.date, self.date - datetime.timedelta(days=1)]:
34            paths[str(date)] = client.list(s3_table_data_path % date)
35
36        completed = set()
37        for date, paths in paths.iteritems():
38            for path in paths:
39                full_path = (s3_table_data_path % date) + path
40                matches = re.findall(r"date=([^/]+)/hour=([0-9]+)", full_path)
41                if len(matches) != 1:
42                    continue
43
44                date, hour = matches[0][0], matches[0][1]
45                if (date, hour) in completed:
46                    continue
47
48                shard_path = (s3_table_data_path % date) + ("hour=%s" % hour)
49                try:
50                    self.query_store("""
51alter table {table} add if not exists partition ( service='drscdn.500px.org', date='{date}', hour={hour} ) location '{path}';
52                    """.format(table=self.table_name, date=date, hour=hour, path=shard_path))
53                except Exception:
54                    pass
55
56                completed.add((date, hour))
57
58
59class QueryViewsByLocation(athena.AthenaQuery):
60    """
61    Query Athena for photo views by location.
62    """
63
64    results_path_template = "fastly-views-by-location/%s"
65
66    def requires(self):
67        return [UpdateFastlyPartitions(date=self.date)]
68
69    def run(self):
70        self.query_store("""
71select date_format(date, '%Y-%m-%d') || ' ' || cast(hour as varchar) || ':00' as date, cast(regexp_extract(path, 'photo/([0-9]+)', 1) as integer) as photo_id, cast(count(*) as integer) as views, round(latitude, 4) as latitude, round(longitude, 4) as longitude from fastly where cast(regexp_extract(path, 'photo/[0-9]+/[^/]*[mhw](?:%%3D|=)([0-9]+)', 1) as bigint) > 600 and response_code in ( 200, 304 ) and date = date '{date}' group by regexp_extract(path, 'photo/([0-9]+)', 1), date, hour, round(latitude, 4), round(longitude, 4)
72        """.format(date=self.date - datetime.timedelta(days=1)))
73
74
75class LoadViewsByLocation(athena.AthenaLoad):
76    """
77    Load photo views by location.
78    """
79
80    table = "fastly_views_by_location"
81
82    def requires(self):
83        return QueryViewsByLocation(date=self.date)
84
85    def create_table(self, connection):
86        connection.cursor().execute("""
87create table {table} (
88  date datetime encode lzo,
89  photo_id int encode lzo,
90  views int encode lzo,
91  latitude decimal(6, 3) encode lzo,
92  longitude decimal(6, 3) encode lzo
93);
94        """.format(table=self.table))
95
96
97if __name__ == '__main__':
98    luigi.run()