<jupyter_start><jupyter_text>Revelio Labs - Data Engineering AssignmentThis is a simplified version of a data engineering project we had at Revelio Labs. The objective is to set up scalable pipelines using the existing stack that can ingest, update and process job posting data from our provider, GHR. You’ll be evaluated on your thought process, your SQL coding skills, how comfortable you are with basic terminal commands and your ability to explain your work. You don’t need to run any code for this exercise. Make sure you’ve read it entirely before you start so that you can see the full picture. Make the best use of online resources to catch up on technologies you are not familiar with. Choose the format that you think is best to gather your answers.Email received from GHR:We're in the process of re-creating an entire image of the dataset for you. It will be placed in a new folder 20200406; the files will be:- master_[index]_[date].csv, (job_id, company, post_date, salary, city) - titles_[index]_[date].csv, (job_id, title)- timelog_[index]_[date].csv, (job_id, remove_date)Each of these files represents a table for our database, and for this run, each table is split across several files (identified by [index]) due to size limitations in AWS.Going forward, new and revised job listing data will be published weekly (typically by Tuesday) in a new folder, dated for the Monday of the week (e.g. next one will be 20200413). These files will be much smaller, since they'll include only a week's worth of new and updated data.About the data:GHR data contains data on job postings starting in 2014. Each job_id is a unique identifier for a job posting (one row per job_id in each table) but sometimes the data is revised by including the same job_id in multiple deliveries. You can find a small sample of each file here (the sets of job_id are not the same among the files): https://info0.s3.us-east-2.amazonaws.com/assignment/engineering/data/master.csv https://info0.s3.us-east-2.amazonaws.com/assignment/engineering/data/title.csv https://info0.s3.us-east-2.amazonaws.com/assignment/engineering/data/timelog.csvSize of the entire image (20200406): master 30Gb, title 6Gb, timelog 9.4GbSize of a weekly update (average): master 1Gb, title 100Mb, timelog 150 MbGHR data (SFTP) Access Details:● Server: ghr-server● UID: client-revelio● private key: stored on EC2 at ~/.ssh/id-rsa-revelio-new● Port: 22About Revelio Labs Stack (AWS)● EC2 r5.4xlarge (remote Ubuntu machine, 16 CPU, 128Gb memory, 9.5Tb disk space).We access it through a terminal on our local machine using the ssh protocol, everything has to be done through the command line once connected to this EC2 machine. You can connect to the GHR bucket using the SFTP protocol on this EC2 machine.https://aws.amazon.com/ec2/● Redshift storage for structured data via PostgreSQL tables (7-node cluster, ~18Tb of disk overall)https://aws.amazon.com/redshift/For the following tasks, please provide commented scripts that perform all the necessary actions. SQL commands should be valid PostgreSQL syntax. Please make the best use of types, keys and indexes. Specify any credentials needed which have not already been specified as “”.  Task 1) SFTP ingestionPlease produce a bash script to transfer the folder YYYYMMDD on the GHR server to the Revelio EC2 instance over SFTP and save the data at “/work/ghr/YYYYMMDD”. This script should work for the entire image of the original dataset and for all following weekly updates. Lets login to the EC2 instance. Hopefully the credentials and iam roles/policies are already set.<jupyter_code>ssh -i <credential.pem> ec2-user@revelio.bunchofawschars.compute-1.amazonaws.com<jupyter_output><empty_output><jupyter_text>Ok lets make it easy on myself. I'm just gonna write a python code that writes a bash. Then we will write a bash that calls that code and the bash to run on the EC2 server.We want to watch out for what we have and what we dont have so we only request gets that are needed. We don't want to redownload things we already have.<jupyter_code>import os 
import datetime as dt
#get current files in directory
existing = set([ name for name in os.listdir('./work/ghr/') if os.path.isdir(os.path.join('./', name)) ])

#get span of dates for folders
folders = set()
start = dt.datetime.strptime("20200403","%Y%m%d")
end = dt.datetime.strptime("20200604","%Y%m%d")
#We can set it to the current time
#end = datetime.today()
week = start
#get all the folder names
while week < end:
    folders.add(week.strftime("%Y%m%d"))
    d = dt.timedelta(days=7)
    week += d
#get the folders we are missing    
getfolders = folders-existing

#here we write the get commands to download the missing folders
with open ('download_files.sh', 'w') as rsh:
    rsh.write('''\
#! /bin/bash\n''')
    for folder in getfolders:
        #note if -i doesnt work, -oIdentityFile=~/.ssh/id-rsa-revelio-new could also work
        rsh.writelines(f'echo "get -r {folder}/ work/ghr/" | sftp -P 22 -i ~/.ssh/id-rsa-revelio-new client-revelio@ghr-server\n')
# here is the bash to download the files
#! /bin/bash
python get_ghr_files.py
bash download_files.sh
#or we can call the bash commands in python instead of writing a whole new bash
import subprocess
for folder in getfolders:
    #note if -i doesnt work, -oIdentityFile=~/.ssh/id-rsa-revelio-new could also work
    command = f'echo "get -r {folder}/ work/ghr/" | sftp -P 22 -i ~/.ssh/id-rsa-revelio-new client-revelio@ghr-server'
    process = subprocess.Popen(command.split(), stdout=subprocess.PIPE)
    output, error = process.communicate()
    print(output, error)<jupyter_output><empty_output><jupyter_text>Task 2) SQL IngestionCreate a single table in PostgreSQL called posting_20200406 containing all the data from each of the three files (job_id, company, post_date, salary, city, title, remove_date) for the initial GHR image.  set up stagingThese are the staging tables we upload the csv files to. The last staging table is the final table we copy the joined staging tables to.<jupyter_code>#We use this code to see the sizes of the string to help determine the datatypes of the table

import pandas as pd
import numpy as np

df = pd.read_csv('title.csv')
measurer = np.vectorize(len)
restitle = measurer(df.values.astype(str)).max(axis=0)

df1 = pd.read_csv('timelog.csv')
restime = measurer(df.values.astype(str)).max(axis=0)


df2 = pd.read_csv('master.csv', encoding='latin-1')
resmaster = measurer(df2.values.astype(str)).max(axis=0)
CREATE TABLE public.stage_master
(
    job_id BIGINT,
    company VARCHAR(200),
    post_date VARCHAR(19),
    salary  INT,
    city VARCHAR(60),
)
DISTSTYLE KEY DISTKEY (job_id) SORTKEY (post_date);

CREATE TABLE public.stage_title
(
    job_id BIGINT,
    title VARCHAR(70),
)
DISTSTYLE KEY DISTKEY (job_id) SORTKEY (title);

CREATE TABLE public.stage_timelog
(
    job_id BIGINT,
    remove_date VARCHAR(19),
)
DISTSTYLE KEY DISTKEY (job_id) SORTKEY (remove_date);
CREATE TABLE public.posting_20200406_stage
(
    job_id BIGINT,
    company VARCHAR(200),
    post_date TIMESTAMP,
    salary  INT,
    city VARCHAR(60),
    title VARCHAR(70),
    remove_date TIMESTAMP,
)
DISTSTYLE KEY DISTKEY (job_id) SORTKEY (remove_date);<jupyter_output><empty_output><jupyter_text>Upload from EC2 to S3 bucketWe could try to upload straight from the EC2 (using a manifest with endpoints and all that jazz), but Redshift is just easier to use with S3 buckets. So lets just copy it over there. Also the cost of the bucket would be like $1 a month so might as well use it. So here is how we can do it with python.We need to make sure the IAM roles and policies are set so the EC2 has s3 access. (Side note: Probably whoever is writing this should have access to EC2 and S3. If not then the user arns should be added so they can modify the EC2 and S3 roles.)<jupyter_code>import boto3

#if bucket doesn't not exist we make one
def create_bucket(session, bucket_name):
    """Creates an AWS S3 bucket in the 'us-east-1' region"""
    try:
        s3_resource = session.resource('s3')
        response = s3_resource.create_bucket(
            Bucket=bucket_name,
            #below is sometimes not needed when the default profile is the same region
            CreateBucketConfiguration={'LocationConstraint': 'us-east-1'}
        )
        
        bucket = s3_resource.Bucket(bucket_name)
        return bucket
        
    except s3_resource.meta.client.exceptions.BucketAlreadyExists:
        bucket = s3_resource.Bucket(bucket_name)
        return bucket
        
    except Exception as e:
        print(e)       
        
def create_EC2_role(session, role_name, policies):
    """Create role with S3 full access policy"""
    try:
        iam_client = session.client('iam')
        policy_document = json.dumps({
            "Version": "2012-10-17",
            "Statement": [
                {
                  "Effect": "Allow",
                  "Principal": {
                    "Service": "ec2.amazonaws.com"
                  },
                  "Action": "sts:AssumeRole"
                }
            ]
        })
        role = iam_client.create_role(
            RoleName=role_name,
            AssumeRolePolicyDocument=policy_document,
        )
        for policy in policies:
            response = iam_client.attach_role_policy(
                RoleName=role_name, 
                PolicyArn=policy
            ) 
        return role    
            
    except Exception as e:
        print(e)

session = boto3.session.Session( 
    aws_access_key_id=<credentials>
    aws_secret_access_key=<credentials>,
    region_name='us-east-1'
)
bucket = (session, 'ghr_data')
role_name = 'EC2toS3'
policies = [
    'arn:aws:iam::aws:policy/AmazonS3FullAccess'
]
role = create_EC2_role(session, role_name, policies)<jupyter_output><empty_output><jupyter_text>Now if the EC2 has AWS CLI installed with it, we can just sync the directories<jupyter_code>! aws s3 sync /work/ghr/ s3://ghr_data<jupyter_output><empty_output><jupyter_text>Otherwise we could just upload it with python & boto3. And while we're here we can use the code to build a manifest for Redshift to read<jupyter_code>import os
files = [f for f in os.listdir('./work/ghr/20200406') if os.path.isfile(f)]
master_entries = []
title_entries = []
timelog_entries = []
for f in files:
    if "csv" in f:
        bucket.upload_file(Filename=f, Key=f'ghr_data/20200406/{f}')
        if "master" in f:
            manifest_master.append({"url":f"s3://ghr_data/20200406/{f}", "mandatory":true})
        elif "title" in f:
            manifest_title.append({"url":f"s3://ghr_data/20200406/{f}", "mandatory":true})
        elif "timelog" in f:
            manifest_timelog.append({"url":f"s3://ghr_data/20200406/{f}", "mandatory":true})
master_manifest = { "entries": master_entries }
title_manifest = { "entries": title_entries }
timelog_manifest = { "entries": timelog_entries }

s3_client = session.client('s3')
putresponse = s3_client.put_object(
     Body=json.dumps(master_manifest),
     Bucket='ghr_data',
     Key='20200406/master.manifest'
)
putresponse = s3_client.put_object(
     Body=json.dumps(title_manifest),
     Bucket='ghr_data',
     Key='20200406/title.manifest'
) 
putresponse = s3_client.put_object(
     Body=json.dumps(timelog_manifest),
     Bucket='ghr_data',
     Key='20200406/timelog.manifest'
)<jupyter_output><empty_output><jupyter_text>copy data to staging tableHere we can run the copy commands using the manifests from S3. First lets add the policies so redshift can access S3. (Side note: Probably whoever is writing this should have access to the Redshift cluster. If not then the user arns should be added so they can modify the red shift roles.)<jupyter_code>def create_redshift_role(session, role_name, policies):
    """Create role with S3 full access policy"""
    try:
        iam_client = session.client('iam')
        policy_document = json.dumps({
            "Version": "2012-10-17",
            "Statement": [
                {
                  "Effect": "Allow",
                  "Principal": {
                    "Service": "redshift.amazonaws.com"
                  },
                  "Action": "sts:AssumeRole"
                }
            ]
        })
        role = iam_client.create_role(
            RoleName=role_name,
            AssumeRolePolicyDocument=policy_document,
        )
        for policy in policies:
            response = iam_client.attach_role_policy(
                RoleName=role_name, 
                PolicyArn=policy
            ) 
        return role    
            
    except Exception as e:
        print(e)
role_name = 'RedshifttoS3'
policies = [
    'arn:aws:iam::aws:policy/AmazonS3FullAccess'
] 
role = create_redshift_role(session, role_name, policies)
# here we upload the files to the tables in Spectrum.
#Hopefully the role has the permissions it needs to use S3
copy public.stage_master
from 's3://ghr_data/20200406/master.manifest' 
iam_role 'arn:aws:iam::10912381340:role/RedshifttoS3'
manifest;
copy public.stage_title
from 's3://ghr_data/20200406/title.manifest' 
iam_role 'arn:aws:iam::10912381340:role/RedshifttoS3'
manifest;
copy public.stage_timelog
from 's3://ghr_data/20200406/timelog.manifest' 
iam_role 'arn:aws:iam::10912381340:role/RedshifttoS3'
manifest;<jupyter_output><empty_output><jupyter_text>copy data from staging to posting_20200406Note we change the date strings to TIMESTAMP and join all the staging tables together<jupyter_code>INSERT into public.posting_20200406_stage
SELECT m.job_id, m.company, TO_TIMESTAMP(m.post_date, 'YYYY-MM-DD HH24:MI:SS') as post_date,
    m.salary, m.city, t.title, TO_TIMESTAMP(tl.remove_date, 'YYYY-MM-DD HH24:MI:SS') as remove_date,
FROM public.stage_master m 
LEFT JOIN public.stage_title t USING(job_id)
LEFT JOIN public.stage_timelog tl USING (job_id);<jupyter_output><empty_output><jupyter_text>Now we need to make sure there are no job duplicates. lets take the job_id with the latest remove_date. Because Redshift uploads in parallel, who knows what file was uploaded when for both the GHR database and this Redshift ingestion? It'd be best if there was a timestamp with the master file of when the records was entered in the GHR database. So lets best guess for duplicates with the lastest removal date as the most recent entry.<jupyter_code>CREATE table public.posting_20200406 DISTSTYLE KEY DISTKEY (job_id) SORTKEY (remove_date) As 
SELECT job_id, company, post_date, salary, city, title, remove_date
FROM
( 
 SELECT *, 
 row_number() over (partition by job_id order by remove_date desc) rno 
 From public.posting_20200406_stage 
) 
where rno = 1;<jupyter_output><empty_output><jupyter_text>Task 3) Data updateNow assume we have a history of posting data received from GHR stored in the tables posting_YYYYMMDD, where YYYYMMDD represents the date the data was received, and a full history of the posting data stored as positng_current. Please explain how you would update the posting_current table with new data stored at posting_20210601.<jupyter_code>#Lets assume that we did well on our part that job_id are distinct on posting_20210601 and posting_current

#We make a copy because deleting is scary
BEGIN;

DROP TABLE IF EXISTS public.posting_current_old
CREATE TABLE public.posting_current_old AS SELECT * FROM public.posting_current;
#for the future we could just truncate the table and insert posting_current

#We delete the jobs from posting current that exist in posting_20210601
DELETE FROM public.posting_current
WHERE job_id IN(
     SELECT job_id
     FROM public.posting_20210601
)
#We insert the posting_20210601 table into public.post_current
INSERT INTO public.posting_current
SELECT *
FROM public.posting_20210601;

COMMIT;<jupyter_output><empty_output><jupyter_text>Task 4) Data processingStarting from the posting_current table, build a table in PostgreSQL that contains the counts of the new, active and removed job postings, as well as the average salaries of the new, active and removed job postings for each month (from 2014 to now). Each row should correspond to a single month and should have the columns (month, count_new, count_active, count_removed, salary_new, salary_active, salary_removed).For a given month, a posting is considered new if it was posted during that month. A posting is considered removed if it was removed during that month. A posting is considered active if it was posted in the current or a previous month and stays active through the month in which it is removed. For example, if posting_current had only one posting in it, posted in 2020-01 and removed in 2020-05, with a salary of 100K, the final aggregated table would look like:<jupyter_code>BEGIN;

# Here we create a new post view for the following fields:
# count_new = posting date month count of job ids
# salary_new = posting date month SUM of salaries
CREATE MATERIALIZED VIEW public.new_post_mv
AS
SELECT TOCHAR(DATE_TRUNC('month', posting_date),'YYYY-MM') AS month,
    COUNT(job_id) AS count_new,
    SUM(salary) AS salary_new
FROM public.posting_current
GROUP BY month;

# Here we create a removed post view for the following fields:
# count_remove = posting removal date count of job ids
# salary_remove = posting removal date SUM of salaries
CREATE MATERIALIZED VIEW public.remove_post_mv
AS
SELECT TO_CHAR(DATE_TRUNC('month', remove_date),'YYYY-MM') AS month,
    COUNT(job_id) AS count_removed,
    SUM(salary) AS salary_removed
FROM public.posting_current
GROUP BY month;

# Here we create a view for the following fields:
# count_active = all dates from posting date to (including) posting removal month count of job ids
# salary_active = all dates from posting date to (including) posting removal month SUM of salaries
# we use generate_series to create dates between the times span of post_date and remove_date\
# Note that we may get nulls in the remove_date because it could still be a current post.
# We could fill the value with NOW(), for this exercise we will used the last date 
CREATE MATERIALIZED VIEW public.active_post_mv
AS
SELECT TO_CHAR(generate_series(DATE_TRUNC('month',post_date),
                               DATE_TRUNC('month',
                                  COALESCE(remove_date,'2020-06-01T00:00:00Z'::TIMESTAMP),
                       '1 month'::interval),
               'YYYY-MM') AS month,
    COUNT(job_id) AS count_active,
    SUM(salary) AS salary_active
FROM public.posting_current
GROUP BY month;

# Here we create a view for the following fields:
# month = all the dates from 2014 to 2020 (NOW() could be used to make it current)
CREATE MATERIALIZED VIEW public.post_dates_mv
AS
SELECT TO_CHAR(generate_series,'YYYY-MM') as month 
FROM generate_series('2014-01-01T00:00:00Z'::TIMESTAMP,
                              '2020-06-01T00:00:00Z'::TIMESTAMP,
                              '1 month'::interval);


# Here we join all the tables on the months from public.post_dates_mv
# We watch out for null for months where there is no data
CREATE TABLE public.post_counts AS 
SELECT pd.month, 
    COALESCE(np.count_new,0) AS count_new,
    COALESCE(ap.count_active,0) AS count_active,
    COALESCE(rp.count_remove,0) AS count_remove,
    COALESCE(np.salary_new,0) AS salary_new,
    COALESCE(ap.salary_active,0) AS salary_active,
    COALESCE(rp.salary_remove,0) AS salary_remove,
FROM public.post_dates_mv pd
LEFT JOIN public.new_post_mv np USING(month)
LEFT JOIN public.remove_post_mv rp USING(month)
LEFT JOIN public.active_post_mv ap USING(month)
ORDER BY month;

COMMIT;<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07978750574653931, 'fraction_numerical': 0.025131531899678195, 'mean_word_length': 3.7856269860669762, 'pattern_counts': {'":': 21, '<': 49, '<?xml version=': 0, '>': 48, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '509717', 'license': 'MIT', 'n_tokens_mistral': 6314, 'n_tokens_neox': 5843, 'n_words': 2195, 'path': 'Data-Ingestion-Redshift/Revelio-1.ipynb', 'repo_name': 'rphillip/Case-Studies', 'size': '19577'}