<jupyter_start><jupyter_text>Assignment_1 AWS Pipeline Architecture-Team 6<jupyter_code># Architectural Components:
# S3 Bucket: https://aws.amazon.com/s3/
# Kinesis Data Stream: https://aws.amazon.com/kinesis/data-streams/
# Kinesis Delivery Stream: https://aws.amazon.com/kinesis/data-firehose/
# Redshift Cluster: https://aws.amazon.com/redshift/
# IAM Policy: https://aws.amazon.com/iam/
# IAM Role: https://aws.amazon.com/iam/<jupyter_output><empty_output><jupyter_text><jupyter_code>!pip install boto3
!pip install psycopg2
#Importing all required libraries:
#pip install boto3
#pip install psycopg2

import boto3
import json
import psycopg2
import datetime
#Setting up the AWS Access keys:
#NOTE: DO NOT EXPOSE THE ACCESS & SECRET KEYS

config_file = open(r'Config.json')
config_file = json.loads(config_file.read())

access_key = config_file['access_key']
secret_access_key = config_file['secret_access_key']

#Configuration Parameters:

bucket_name = "info7374s3assignment2"#------------------------------------S3: Name of the creating AWS S3 Bucket

policy_name = "info7374policyassignment2"#----------------------------------IAM: Access policy name
iam_role_name = "info7374roleassignment2"#----------------------------------IAM: IAM Role for the architecural access

db_name = "info7374dbassignment2"#-------------------------------------Redshift: Database Name for gaming data
cluster_identifier = "info7374clusterassignment2"#---------------------Redshift: Redshift Cluster Name
master_username = "root"#----------------------------------------Redshift: Admin Username
master_password = "<PASSWORD>"#---------------------------------Redshift: Admin Password
node_type = "dc2.large"#-----------------------------------------Redshift: Cluster configuration
cluster_type = "single-node"#------------------------------------Redshift: Cluster Type
availability_zone = "us-east-1a"#--------------------------------Redshift: Cluster Availability Zone
table_name = "info7374_game"#------------------------------------Redshift: Database table name
kinesis_data_stream_name = "info7374datastreamassignment2"#-------------Kinesis: Data Stream Name
shard_count = 100#------------------------------------------------Kinesis: Data Stream Shard Count

log_group_name = "info7374loggroupassignment2"#------------------------CloudWatch: Log Group Name
log_stream_name = "info7374logstreamassignment2"#------------------------CloudWatch: Log Group Name
delivery_stream_name = "info7374deliverystreamassignment2"#--------------Kinesis: Delivery Stream Name
stream_type = "KinesisStreamAsSource"#-----------------------------Kinesis: Delivery Stream Type

#Creating AWS S3 Bucket:

s3_client = boto3.client('s3', aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)

response_s3 = s3_client.create_bucket(ACL="private", Bucket=bucket_name)
print(response_s3)
bucket_arn = "arn:aws:s3:::"+bucket_name#---------------------------S3: Bucker ARN
print(bucket_arn)
#Creating IAM Policy for Architecural access:

iam_client = boto3.client('iam', aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)


policy_details = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "glue:GetTableVersions"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:GetFunctionConfiguration"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kms:Decrypt"
            ],
            "Resource": "*"
        }
    ]
}
response_iam_policy = iam_client.create_policy(PolicyName=policy_name, PolicyDocument=json.dumps(policy_details))
print(response_iam_policy)
policy_arn = response_iam_policy['Policy']['Arn']#------------------------IAM: Policy ID for assignment
print(policy_arn)

#Creating IAM Role for Architectural access:

assume_role_policy_doc = {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "firehose.amazonaws.com"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "237888173777"
        }
      }
       
    }
  ]
}

response_iam_role = iam_client.create_role(RoleName=iam_role_name, AssumeRolePolicyDocument=json.dumps(assume_role_policy_doc))

print(response_iam_role)
role_arn = response_iam_role['Role']['Arn']#----------------------------IAM: Role ARN
#Attaching a Policy to a Role:

response_iam_role_policy_attach = iam_client.attach_role_policy(PolicyArn=policy_arn,RoleName=iam_role_name)
print(response_iam_role_policy_attach)

#Creating AWS Redshift Cluster:

redshift_client = boto3.client('redshift',region_name="us-east-1", aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)

response_redshift = redshift_client.create_cluster(
    DBName=db_name,
    ClusterIdentifier=cluster_identifier,
    ClusterType=cluster_type,
    NodeType=node_type,
    MasterUsername=master_username,
    MasterUserPassword=<PASSWORD>,
    AvailabilityZone=availability_zone,
    IamRoles=[role_arn],
    PubliclyAccessible=True)

print(response_redshift)
#Note: Wait for 5 min to get AWS Redshift cluster to setup as we need to create a Database table.

#Describe AWS Redshift Cluster to get an endpoint:

response_redshift_desc = redshift_client.describe_clusters(ClusterIdentifier=cluster_identifier)
print(response_redshift_desc)
hostname = response_redshift_desc['Clusters'][0]['Endpoint']['Address']#----------------Redshift: Hostname for database
port_number = response_redshift_desc['Clusters'][0]['Endpoint']['Port']#----------------Redshift: Port Number for databse
#Creating Database table on Redshift:

con = psycopg2.connect(dbname= db_name, host= hostname, port= port_number, user= master_username, password= <PASSWORD>)

con.set_session(autocommit=True)
cur = con.cursor()
#Example: https://docs.aws.amazon.com/redshift/latest/dg/r_CREATE_TABLE_examples.html

query = "create table Products (\n"
query+= "product_id varchar(200) not null,\n"
query+= "product_name varchar(200) not null,\n"
query+= "category varchar(200) not null);"

print(query)
cur.execute(query)

query = "create table sales (\n"
query+= "Order_id integer not null,\n"
query+= "Customer_ID varchar(200) not null,\n"
query+= "Customer_Name varchar(200) not null,\n"
query+= "start_date date not null,\n"
query+= "end_date date not null,\n"
query+= "product_id varchar(200) not null,\n"
query+= "Price decimal(8,2) not null,\n"
query+= "quantity integer not null,\n"
query+= "total_amount integer not null,\n"
query+= "Salesperson_id varchar(200) not null);"

print(query)
cur.execute(query)
query = "create table Customers (\n"
query+= "Customer_ID integer not null,\n"
query+= "Company_Name varchar(200) not null,\n"
query+= "City varchar(200) not null,\n"
query+= "Country varchar(200) not null,\n"
query+= "State varchar(200) not null,\n"
query+= "Company_Strength integer not null,\n"
query+= "Company_Revenue decimal(10,3) not null,\n"
query+= "Estd_before_2009 varchar(200) not null,\n"
query+= "Decline_AMC_Contract varchar(200) not null,\n"
query+= "Lead_Source varchar(200) not null,\n"
query+= "Repeat_Customer varchar(200) not null,\n"
query+= "Roadshow varchar(200) not null,\n"
query+= "Phone_call varchar(200) not null,\n"
query+= "Brochures varchar(200) not null,\n"
query+= "Payment_default_in_last_5years varchar(200) not null,\n"
query+= "Converted boolean not null);"

print(query)


cur.execute(query)
query = "create table Products_pricing (\n"
query+= "product_id varchar(200) not null,\n"
query+= "Making_Cost  decimal(8,2) not null,\n"
query+= "Delivery_Charges decimal(8,2) not null,\n"
query+= "Overhead_cost decimal(8,2) not null,\n"
query+= "cost_price decimal(8,2) not null,\n"
query+= "selling_price decimal(8,2) not null);"

print(query)

cur.execute(query)
query = "create table Promotions (\n"
query+= "Promotion_id varchar(200) not null,\n"
query+= "Promotion_Name varchar(200) not null,\n"
query+= "Promotion_Start_Date date not null,\n"
query+= "Promotion_End_Date date not null);"

print(query)
cur.execute(query)

query = "create table new_Customers (\n"
query+= "Customer_ID integer not null,\n"
query+= "Company_Name varchar(200) not null,\n"
query+= "City varchar(200) not null,\n"
query+= "Country varchar(200) not null,\n"
query+= "State varchar(200) not null,\n"
query+= "Company_Strength integer not null,\n"
query+= "Company_Revenue decimal(10,3) not null,\n"
query+= "Estd_before_2009 varchar(200) not null,\n"
query+= "Decline_AMC_Contract varchar(200) not null,\n"
query+= "Lead_Source varchar(200) not null,\n"
query+= "Repeat_Customer varchar(200) not null,\n"
query+= "Roadshow varchar(200) not null,\n"
query+= "Phone_call varchar(200) not null,\n"
query+= "Brochures varchar(200) not null,\n"
query+= "Payment_default_in_last_5years varchar(200) not null,\n"
query+= "Converted boolean not null);"

print(query)


cur.execute(query)

query = "create table Test_data (\n"
query+= "Customer_ID integer not null,\n"
query+= "Company_Name varchar(200) not null,\n"
query+= "City varchar(200) not null,\n"
query+= "Country varchar(200) not null,\n"
query+= "State varchar(200) not null,\n"
query+= "Company_Strength integer not null,\n"
query+= "Company_Revenue decimal(10,3) not null,\n"
query+= "Estd_before_2009 varchar(200) not null,\n"
query+= "Decline_AMC_Contract varchar(200) not null,\n"
query+= "Lead_Source varchar(200) not null,\n"
query+= "Repeat_Customer varchar(200) not null,\n"
query+= "Phone_call varchar(200) not null,\n"
query+= "Payment_default_in_last_5years varchar(200) not null,\n"
query+= "Converted boolean not null,\n"
query+= "lead_source_Dealer integer not null,\n"
query+= "lead_source_Email integer not null,\n"
query+= "lead_source_RFQ integer not null,\n"
query+= "lead_source_Telephonic integer not null,\n"
query+= "roadshow_Attended integer not null,\n"
query+= "roadshow_Attended_Multipl_events integer not null,\n"
query+= "roadshow_Registered integer not null,\n"
query+= "brochures_Downloaded integer not null,\n"
query+= "brochures_Viewed integer not null,\n"
query+= "brochures_Viewed_multiple_times integer not null);"


print(query)
#Creating Kinesis Stream:

kinesis_client = boto3.client('kinesis',region_name="us-east-1", aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)

response_kinesis_data_stream = kinesis_client.create_stream(StreamName=kinesis_data_stream_name,ShardCount=shard_count)
print(response_kinesis_data_stream)
response_kinesis_data_stream_desc = kinesis_client.describe_stream(StreamName=kinesis_data_stream_name)
print(response_kinesis_data_stream_desc)
kinesis_stream_arn = response_kinesis_data_stream_desc['StreamDescription']['StreamARN']#-----------------------Kinesis: Datastream ARN
print(kinesis_stream_arn)

#Creating Kinesis Delivery Stream: Firehose

firehose_client = boto3.client('firehose',region_name="us-east-1", aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)

cluster_jdbc_url = "jdbc:redshift://"+hostname+":"+str(port_number)+"/"+db_name
print(cluster_jdbc_url)
response_firehose = firehose_client.create_delivery_stream(
    DeliveryStreamName = delivery_stream_name,
    DeliveryStreamType = stream_type,
    KinesisStreamSourceConfiguration={
        'KinesisStreamARN': kinesis_stream_arn,
        'RoleARN': 'arn:aws:iam::237888173777:role/info7374roleassignment2'
    },
    RedshiftDestinationConfiguration={
        'RoleARN': 'arn:aws:iam::237888173777:role/info7374roleassignment2',
        'ClusterJDBCURL': cluster_jdbc_url,
        'CopyCommand': {
            'DataTableName': 'test_data',
            
            'DataTableColumns': 'customer_id,company_name,city,country,state,company_strength,company_revenue,estd_before_2009,decline_amc_contract,lead_source,repeat_customer,phone_call,payment_default_in_last_5years,converted,lead_source_dealer,lead_source_email,lead_source_rfq,lead_source_telephonic,roadshow_attended,roadshow_attended_multipl_events,roadshow_registered,brochures_downloaded,brochures_viewed,brochures_viewed_multiple',
            'CopyOptions': "json 'auto'"
        },
        'Username': master_username,
        'Password': <PASSWORD>,
        'S3Configuration': {
            'RoleARN': 'arn:aws:iam::237888173777:role/info7374roleassignment2',
            'BucketARN': 'arn:aws:s3:::info7374s3assignment2',
            'BufferingHints': {
                'IntervalInSeconds': 60
            }
        }
    })

print(response_firehose)

#Cogratulation!!! At this stage we have successfully created our Architacure
#Real-time data streaming:

kinesis_delivery_client = boto3.client('kinesis', region_name="us-east-1", aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)

#My Data

import pandas as pd


#Reading the csv file
data_file = r'test_data.csv'
data = pd.read_csv(data_file)
data.head()
def set_default(obj):
  if isinstance(obj, set):
      return list(obj)
  raise TypeError
for i in range(0,data.shape[0]):

  customer_id=int(data['customer_id'][i])
  company_name= data['company_name'][i]
  city=data['city'][i]
  country=data['country'][i]
  state=data['state'][i]
  company_strength=int(data['company_strength'][i])
  company_revenue=int(data['company_revenue'][i])
  estd_before_2009=str(data['estd_before_2009'][i])
  decline_amc_contract=str(data['decline_amc_contract'][i])
  repeat_customer=str(data['repeat_customer'][i])
  phone_call=data['phone_call'][i]
  payment_default_in_last_5years=str(data['payment_default_in_last_5years'][i])
  converted=int(data['converted'][i])
  lead_source_dealer=int(data['lead_source_Dealer'][i])
  lead_source_email=int(data['lead_source_Email'][i])
  lead_source_rfq=int(data['lead_source_RFQ'][i])
  lead_source_telephonic=int(data['lead_source_Telephonic'][i])
  roadshow_attended=int(data['roadshow_Attended'][i])
  roadshow_attended_multipl_events=int(data['roadshow_Attended Multiple events'][i])
  roadshow_registered=int(data['roadshow_Registered'][i])
  brochures_downloaded=int(data['brochures_Downloaded'][i])
  brochures_viewed=int(data['brochures_Viewed'][i])
  brochures_viewed_multiple=int(data['brochures_Viewed multiple times'][i])

  data_row = {"customer_id":customer_id,"company_name":company_name,"city":city,"country":country,"state":state,"company_strength":company_strength,"company_revenue":company_revenue,"estd_before_2009":estd_before_2009,"decline_amc_contract":decline_amc_contract,"repeat_customer":repeat_customer,"phone_call":phone_call,"payment_default_in_last_5years":payment_default_in_last_5years,"converted":converted,'lead_source_Dealer':lead_source_dealer,'lead_source_Email':lead_source_email, 'lead_source_RFQ':lead_source_rfq, 'lead_source_Telephonic':lead_source_telephonic,'roadshow_Attended':roadshow_attended, 'roadshow_Attended Multiple events':roadshow_attended_multipl_events,'roadshow_Registered':roadshow_registered, 'brochures_Downloaded':brochures_downloaded, 'brochures_Viewed':brochures_viewed,'brochures_Viewed multiple times':brochures_viewed_multiple}
  data_row = json.dumps(data_row)

  data_row = bytes(data_row, 'utf-8')
  formatted_records = [{'PartitionKey': "pk",'Data': data_row}]
  print(formatted_records)
  delivery_stream_name = kinesis_data_stream_name
  response = kinesis_delivery_client.put_records(StreamName=delivery_stream_name, Records=formatted_records)
  break
response



#Now we will demolish the complete architecture:

#Delete Redshift Cluster:
response_delete_redshift = redshift_client.delete_cluster(
    ClusterIdentifier=cluster_identifier,
    SkipFinalClusterSnapshot=True
)

print(response_delete_redshift)
#Delete Kinesis Delivery stream:
response_delete_firehose = firehose_client.delete_delivery_stream(DeliveryStreamName=delivery_stream_name)
print(response_delete_firehose)
#Delete Kinesis Data stream:
response_delete_data_stream = kinesis_client.delete_stream(StreamName=kinesis_data_stream_name,EnforceConsumerDeletion=True)
print(response_delete_data_stream)
#Delete S3 Bucket:

#All of the Objects in a bucket must be deleted before the bucket itself can be deleted:

s3 = boto3.resource('s3', aws_access_key_id = access_key, aws_secret_access_key = secret_access_key)
bucket = s3.Bucket(bucket_name)

for key in bucket.objects.all():
    key.delete()
bucket.delete()

print("Bucket deleted successfully!")
#Delete IAM Policy:


#Delete IAM Role:

response_detach_policy = iam_client.detach_role_policy(PolicyArn=policy_arn,RoleName=iam_role_name)
print(response_detach_policy)

response_delete_role = iam_client.delete_role(RoleName=iam_role_name)
print(response_delete_role)
#Delete IAM Policy:

response_delete_policy = iam_client.delete_policy(PolicyArn=policy_arn)
print(response_delete_policy)
s3 = boto3.resource('s3')
bucket = s3.Bucket('info7374s3assignment2')
s3.Object('info7374s3assignment2', 'hello.txt').put(Body=open('/tmp/hello.txt', 'rb'))
import os

path = '2nd assignment dma'

files = []
# r=root, d=directories, f = files
for r, d, f in os.walk(path):
    for file in f:
        if '.txt' in file:
            files.append(os.path.join(r, file))

for f in files:
    print(f)
import psycopg2
con=psycopg2.connect(dbname= db_name, host=hostname, 
port=port_number, user=master_username, password= <PASSWORD>)
cur = con.cursor()
cur.execute("SELECT * FROM customers;")

cur.fetchall()
pip install pandas_redshift
import numpy as np
data = np.array(cur.fetchall())
import pandas_redshift as pr
pip install pandas_redshift
import pandas_redshift as pr
pr.connect_to_redshift(dbname = db_name ,
                        host = hostname,
                        port = port_number,
                        user = master_username,
                        password =<PASSWORD>)
data = pr.redshift_to_pandas('select * from customers')
data
data['company_revenue'] = data['company_revenue'].convert_objects(convert_numeric=True)
data.dtypes
multi1={"TRUE":1,"FALSE":0}
data.repeat_customer=[multi1[item] for item in data.repeat_customer]
data
multi1={"TRUE":1,"FALSE":0}
data.payment_default_in_last_5years=[multi1[item] for item in data.payment_default_in_last_5years]
data
import pandas as pd
data=pd.get_dummies(data, columns=["lead_source"])
data.head()
data=pd.get_dummies(data, columns=["roadshow"])
data.head()
data=pd.get_dummies(data, columns=["brochures"])
data.head()
multi1={"TRUE":1,"FALSE":0}
data.estd_before_2009=[multi1[item] for item in data.estd_before_2009]
data
multi1={"TRUE":1,"FALSE":0}
data.decline_amc_contract=[multi1[item] for item in data.decline_amc_contract]
data
data
data['converted'] = (data['converted'] == True ).astype(int)
data
data.corr()

import matplotlib.pyplot as plt
import seaborn as sns

plt.figure(figsize=(16,12))
sns.heatmap(data=data.corr(),annot=True,fmt='.2f',cmap='coolwarm')
plt.show()
data_data=data.ix[:,(5,6,7,8,9,12,13,14,15,16,16,17,18,19,20,21)].values
y=data.ix[:,(11)].values
x=scale(data_data)
LogReg=LogisticRegression()
LogReg.fit(x,y)
#print (LogReg.score(x,y))
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
from scipy import stats
import seaborn as sns
import statsmodels.api as sm
from sklearn import linear_model
from statsmodels.formula.api import ols
from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.preprocessing import PolynomialFeatures
from sklearn.pipeline import make_pipeline

import warnings
warnings.filterwarnings('ignore')
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.tree import DecisionTreeRegressor
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error, mean_squared_log_error
from sklearn.preprocessing import scale
from sklearn.linear_model import LogisticRegression
from sklearn import metrics
from sklearn import preprocessing
from pylab import rcParams
from scipy.stats import spearmanr
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.preprocessing import Imputer
from statsmodels.stats.outliers_influence import variance_inflation_factor


from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import SGDClassifier
from sklearn.datasets import load_iris
from sklearn.metrics import classification_report
#Reading the csv file
new_data_file = r'test_data.csv'
new_data = pd.read_csv(new_data_file)
new_data
new_data['company_name']

new_data_ch=new_data.ix[:,(5,6,7,8,9,12,13,14,15,16,16,17,18,19,20,21)].values
#y=data.ix[:,(11)].values
c=scale(new_data_ch)
e=LogReg.predict_proba(c)

i=0
yes_array=[]
for value in e:
  print(e[i,0])
  yes_array.append(e[i,0])
  i+=1
  

df_toplot=pd.DataFrame({
    "probability for customer" : yes_array,
    "customer" : new_data['company_name']
    
})

df_toplot
xaxis=df_toplot['probability for customer']
yaxis=df_toplot['customer']
plt.bar(yaxis,xaxis)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14662437698232894, 'fraction_numerical': 0.022564567285908475, 'mean_word_length': 4.673778920308483, 'pattern_counts': {'":': 57, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '624568', 'license': 'MIT', 'n_tokens_mistral': 7590, 'n_tokens_neox': 7186, 'n_words': 1405, 'path': 'Assignment 2/Assignment_2_team6.ipynb', 'repo_name': 'purvangt25/INFO7374-Digital-Marketing-Analytics', 'size': '22096'}