uploading workload execution files

Author: kevinassogba

README.md

@@ -0,0 +1 @@
+# canary

datamining/divindex.py

@@ -0,0 +1,16 @@
+import subprocess
+import time
+
+def main(args):
+    initTime = time.time()
+    invocation = "java -jar /USCensus.jar US_diversity_index"
+    __post(invocation)
+    print("INFO ::> Job Completed @ {0} seconds".format(time.time()-initTime))
+    return {"Status": "Job completed"}
+
+
+def __post(command):
+    response = subprocess.run(['/bin/bash', '-c', command], stdout=subprocess.PIPE,
+                              stderr=subprocess.STDOUT).stdout.decode().strip()
+    print('Result:', response)
+

datamining/usdivindex.log

@@ -0,0 +1,88 @@
+import sys
+import psycopg2
+import json
+from psycopg2.extras import Json
+# Redundant function
+# Used to find nested json keys
+def id_generator(dict_var):
+    for k, v in dict_var.items():
+        v = isinstance(v, str) and v.replace("True", "true").replace("False", "false").replace('"', "'").replace("'", '"')
+        try:
+            v = json.loads(v)
+            print(f'Parsed: {k}: {v}')
+        except:
+            print(f'Cannot parse {k}: {v}')
+            pass
+            # Dont show any error
+        if isinstance(v, dict):
+            for id_val in id_generator(v):
+                yield id_val
+        else:
+            yield k
+def process_business_attributes():
+    conn = psycopg2.connect(host='172.17.0.6', user='postgres', dbname='canary_db')
+    cur = conn.cursor()
+    cur.execute("SELECT * FROM business_attribute")
+    rows = cur.fetchall()
+    ops = []
+    for i, r in enumerate(rows):
+        isUpdated = False
+        attributes = r[1]
+        for k, v in attributes.items():
+            if isinstance(v, str) and '{' in v and '}' in v:
+                # Replace True and False to true and false respectively as these are the standard JSON notations.
+                # Replace the single quotes with double and the double ones with single, this is a problem in the source data. JSON data needs to have the keys enclosed in double quotes.
+                v = v.replace("True", "true").replace("False", "false").replace('"', "'").replace("'", '"')  
+                try:
+                    v = json.loads(v)
+                    attributes[k] = v
+                    isUpdated = True
+                except:
+                    pass
+        if isUpdated:
+            ops.append({"business_id": r[0], "attributes": Json(attributes)})
+    print(f'Starting to update {len(ops)} rows in business_attribute table...')
+    # The following query takes around 50 minutes
+    cur.executemany("UPDATE business_attribute SET attribute = %(attributes)s WHERE business_id = %(business_id)s", ops)
+    print(f'Updated {len(ops)} rows in business_attributes')
+    # This query takes around 13 seconds
+    cur.execute(
+            """
+            CREATE TABLE business_attribute_temp AS (with recursive flat(business_id, key, value) as
+            (
+                SELECT business_id, key, value FROM business_attribute, jsonb_each(attribute)
+                UNION
+                SELECT f.business_id, concat(f.key, '.', j.key), j.value FROM flat f, jsonb_each(f.value) j WHERE jsonb_typeof(f.value) = 'object'
+            )
+            SELECT business_id, jsonb_object_agg(key, value) as data from flat WHERE jsonb_typeof(value) <> 'object' GROUP BY business_id);
+            create or replace function create_jsonb_flat_view
+                (table_name text, regular_columns text, json_column text)
+                returns text language plpgsql as $$
+            declare
+                cols text;
+            begin
+                execute format ($ex$
+                    select string_agg(format('%2$s->>%%1$L "%%1$s"', key), ', ')
+                    from (
+                        select distinct key
+                        from %1$s, jsonb_each(%2$s)
+                        order by 1
+                        ) s;
+                    $ex$, table_name, json_column)
+                into cols;
+                execute format($ex$
+                    drop view if exists %1$s_view;
+                    create view %1$s_view as 
+                    select %2$s, %3$s from %1$s
+                    $ex$, table_name, regular_columns, cols);
+                return cols;
+            end $$;
+            SELECT create_jsonb_flat_view('business_attribute_temp', 'business_id', 'data');
+            CREATE TABLE business_attributes_exploded AS (SELECT * FROM business_attribute_temp_view);
+	    ALTER TABLE business_attributes_exploded ADD PRIMARY KEY (business_id);
+            """
+            )
+    cur.close()
+    conn.commit()
+
+process_business_attributes()

dbquery/dbquery.log

@@ -0,0 +1,44 @@
+#!/usr/bin/python3
+#!/usr/bin/env python3
+import psycopg2
+import subprocess, time
+from psycopg2.extras import Json
+
+def main(args):
+    initTime = time.time()
+    pg_hostname = subprocess.run(['/bin/bash', '-c', "docker inspect -f '{{range.NetworkSettings.Networks}}{{.IPAddress}}{{end}}' canary-mongodb"], stdout=subprocess.PIPE,stderr=subprocess.PIPE,universal_newlines=True)
+    pg_username = "postgres"
+    pg_password = "password"
+    pg_db = "canary_db"
+    #conn = psycopg2.connect(f'host={pg_hostname} user={pg_username} dbname={pg_db}')
+    conn = psycopg2.connect(host='172.17.0.6', user='postgres', dbname='canary_db')
+    cur = conn.cursor()
+    print("CANARY ::> Connection Establish -> Time = {0}.s".format(time.time()-initTime))
+
+    for state in range(100):
+        print("CANARY ::> Iteration {0} starting -> Time = {1}.s".format(state, time.time()-initTime))
+        cur.execute("select t1.business_id, t1.name, t1.city, t1.state, t1.stars, t1.review_count, count(t2.id) as checkins, sum(t3.cool) as cool_reviews from business_info t1 JOIN checkin_info t2 on t1.business_id=t2.business_id JOIN review_info t3 on t2.business_id=t3.business_id where t1.city ilike 'Pittsburgh' group by t1.business_id order by checkins desc limit 10;")
+        print("CANARY ::> Query 1 completed -> Time = {0}.s".format(time.time()-initTime))
+
+        cur.execute("select t1.business_id, t1.name, t1.city, t1.address, t1.review_count, t2.closing_hour, t3.category, t5.stars from business_info t1 JOIN business_hour t2 on t1.business_id=t2.business_id JOIN business_category t3 on t2.business_id=t3.business_id JOIN review_info t5 on t3.business_id=t5.business_id JOIN business_attributes_exploded t4 on t5.business_id=t4.business_id where t1.city='Las Vegas' AND t2.closing_hour > '22:00:00' AND t3.category='Restaurants' AND t5.stars > 3 AND t4.\"GoodForKids\" ilike 'True';")
+        print("CANARY ::> Query 2 completed -> Time = {0}.s".format(time.time()-initTime))
+
+        cur.execute("select t1.business_id, t1.name, t1.city, t1.state, t1.address, t1.stars, t2.category from business_info t1 JOIN business_category t2 on t1.business_id=t2.business_id where t1.stars>4 AND t1.state='NY' AND t2.category='Sandwiches' limit 50;")
+        print("CANARY ::> Query 3 completed -> Time = {0}.s".format(time.time()-initTime))
+
+        cur.execute("select t1.business_id, t1.name, t1.city, t1.state, t1.address, t1.stars, t1.review_count, t3.category from business_info t1 JOIN review_info t2 on t1.business_id=t2.business_id JOIN business_category t3 on t2.business_id=t3.business_id where t2.date > '2005-01-01 00:00:00' AND t2.date < '2006-01-01 00:00:00' AND t3.category ILIKE '%resort%' GROUP BY t1.business_id, t3.category ORDER BY t1.review_count DESC limit 50;")
+        print("CANARY ::> Query 4 completed -> Time = {0}.s".format(time.time()-initTime))
+
+        cur.execute("select t1.business_id, t1.name, t1.city, t1.state, t1.stars, t1.review_count from business_info t1 where is_open=0 and stars <2 limit 10;")
+        print("CANARY ::> Query 5 completed -> Time = {0}.s".format(time.time()-initTime))
+
+        cur.execute("select t1.city, t1.state, count(t2.review_id) as total_reviews from business_info t1 JOIN review_info t2 on t1.business_id=t2.business_id group by t1.city,t1.state order by total_reviews desc limit 10; ")
+        print("CANARY ::> Query 6 completed -> Time = {0}.s".format(time.time()-initTime))
+        print("CANARY ::> State {0} collection -> Time = {1}.s".format(state, time.time()-initTime))
+
+    cur.close()
+    conn.commit()
+    print("CANARY ::> Job completed -> Time = {0}.s".format(time.time()-initTime))
+
+    return {'CANARY STATUS': "Queries completed successfully."}
+