Skip to content
This repository has been archived by the owner. It is now read-only.
Permalink
Browse files
Merge pull request #23 from radibnia77/main
Update dlpredictor pipeline
  • Loading branch information
radibnia77 committed Jan 4, 2022
2 parents 6a03f7a + 3f574a4 commit 0e97e79d2caf59f2228725367d544e4ac78dc34d
Showing 4 changed files with 529 additions and 1 deletion.
@@ -181,6 +181,9 @@ def run(hive_context, cluster_size_cfg, input_table_name,
df = df.withColumn('imp', udf(lambda ts: sum(
[_ for _ in ts if _]), IntegerType())(df.ts))

# remove uckeys with 0 imp
df = df.filter('imp>0')

# add popularity = mean
df = df.withColumn('p', udf(lambda ts: sum(
[_ for _ in ts if _])/(1.0 * len(ts)), FloatType())(df.ts))
@@ -52,7 +52,7 @@ def run(hive_context, input_table_name, outlier_table):
def _filter_outlier(x, ind_list):
for i in range(1, len(x)-1):
if i in ind_list and x[i] != None and x[i + 1] != None and x[i - 1] != None:
x[i] = (x[i - 1] + x[i + 1]) / 2
x[i] = int((x[i - 1] + x[i + 1]) / 2)
return x

command = """
@@ -0,0 +1,270 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at

# http://www.apache.org/licenses/LICENSE-2.0.html

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import unittest
import math
import pickle
import statistics
import yaml
import argparse
import re
import hashlib

import pyspark.sql.functions as fn
import numpy as np

from pyspark import SparkContext
from pyspark.sql import SparkSession, HiveContext
from pyspark.sql.types import IntegerType, StringType, MapType
from datetime import datetime, timedelta

'''
This file operates like check_model but only produces the output, no verification.
This script performs the following actions:
1. call model API with N number of randomly picked dense uckeys from trainready (The same data that is used to train the model).
2. calculate the accuracy of the model.
run by:
spark-submit --master yarn --num-executors 5 --executor-cores 3 --executor-memory 16G --driver-memory 16G get_model_diff.py
'''

from client_rest_dl2 import predict


def c_error(x, y):
x = x * 1.0
if x != 0:
e = abs(x - y) / x
else:
e = -1
e = round(e, 3)
return e


def error_m(a, p):
result = []
for i in range(len(a)):
x = a[i]
y = p[i]
e = c_error(x, y)
result.append(e)
x = sum(a)
y = sum(p)
e = c_error(x, y)
return (e, result)


def normalize_ts(ts):
ts_n = [math.log(i + 1) for i in ts]
return ts_n


def dl_daily_forecast(serving_url, model_stats, day_list, ucdoc_attribute_map):
x, y = predict(serving_url=serving_url, model_stats=model_stats,
day_list=day_list, ucdoc_attribute_map=ucdoc_attribute_map, forward_offset=0)
ts = x[0]
days = y
return ts, days


def get_model_stats(hive_context, model_stat_table):
'''
return a dict
model_stats = {
"model": {
"name": "s32",
"version": 1,
"duration": 90,
"train_window": 60,
"predict_window": 10
},
"stats": {
"g_g_m": [
0.32095959595959594,
0.4668649491714752
],
"g_g_f": [
0.3654040404040404,
0.4815635452904544
],
"g_g_x": [
0.31363636363636366,
0.46398999646418304
],
'''
command = """
SELECT * FROM {}
""".format(model_stat_table)
df = hive_context.sql(command)
rows = df.collect()
if len(rows) != 1:
raise Exception('Bad model stat table {} '.format(model_stat_table))
model_info = rows[0]['model_info']
model_stats = rows[0]['stats']
result = {
'model': model_info,
'stats': model_stats
}
return result


def predict_daily_uckey(sample, days, serving_url, model_stats, columns):

def _denoise(ts):
non_zero_ts = [_ for _ in ts if _ != 0]
nonzero_p = 0.0
if len(non_zero_ts) > 0:
nonzero_p = 1.0 * sum(ts) / len(non_zero_ts)

return [i if i > (nonzero_p / 10.0) else 0 for i in ts]

def _helper(cols):
day_list = days[:]
ucdoc_attribute_map = {}
for feature in columns:
ucdoc_attribute_map[feature] = cols[feature]

# determine ts_n and days
model_input_ts = []

# -----------------------------------------------------------------------------------------------
'''
The following code is in dlpredictor, here ts has a different format
'ts': [0, 0, 0, 0, 0, 65, 47, 10, 52, 58, 27, 55, 23, 44, 38, 42, 90, 26, 95, 34, 25, 26, 18, 66, 31,
0, 38, 26, 30, 49, 35, 61, 0, 55, 23, 44, 35, 33, 22, 25, 28, 72, 25, 15, 29, 29, 9, 32, 18, 20, 70,
20, 4, 11, 15, 10, 8, 3, 0, 5, 3, 0, 23, 11, 44, 11, 11, 8, 3, 38, 3, 28, 16, 3, 4, 20, 5, 4, 45, 15, 9, 3, 60, 27, 15, 17, 5, 6, 0, 7, 12, 0],
# ts = {u'2019-11-02': [u'1:862', u'3:49', u'2:1154'], u'2019-11-03': [u'1:596', u'3:67', u'2:1024']}
ts = ucdoc_attribute_map['ts'][0]
price_cat = ucdoc_attribute_map['price_cat']
for day in day_list:
imp = 0.0
if day in ts:
count_array = ts[day]
for i in count_array:
parts = i.split(':')
if parts[0] == price_cat:
imp = float(parts[1])
break
model_input_ts.append(imp)
'''
model_input_ts = ucdoc_attribute_map['ts']
price_cat = ucdoc_attribute_map['price_cat']

# --------------------------------------------------------------------------------------------------------

# remove science 06/21/2021
# model_input_ts = replace_with_median(model_input_ts)

model_input_ts = _denoise(model_input_ts)

ts_n = normalize_ts(model_input_ts)
ucdoc_attribute_map['ts_n'] = ts_n

# add page_ix
page_ix = ucdoc_attribute_map['uckey'] + '-' + price_cat
ucdoc_attribute_map['page_ix'] = page_ix

rs_ts, rs_days = dl_daily_forecast(
serving_url=serving_url, model_stats=model_stats, day_list=day_list, ucdoc_attribute_map=ucdoc_attribute_map)

# respose = {'2019-11-02': 220.0, '2019-11-03': 305.0}

response = {}
for i, day in enumerate(rs_days):
response[day] = rs_ts[i]
return response

return _helper(cols=sample)


def run(cfg, cfg_1, hive_context):

model_stats = get_model_stats(hive_context, cfg['model_stat_table'])
model_stats_1 = get_model_stats(hive_context, cfg_1['model_stat_table'])

# create day_list from yesterday for train_window
duration = model_stats['model']['duration']
predict_window = model_stats['model']['predict_window']
day_list = model_stats['model']['days']
day_list.sort()

local = False
if not local:
df_trainready = hive_context.sql(
'SELECT * FROM {} '.format(cfg['trainready_table']))
df_dist = hive_context.sql(
'SELECT * FROM {} WHERE ratio=1'.format(cfg['dist_table']))
df = df_trainready.join(
df_dist, on=['uckey', 'price_cat'], how='inner')
columns = df.columns
samples = df.take(cfg['max_calls'])

errs = []
for _ in samples:
sample = {}
for feature in columns:
sample[feature] = _[feature]

sample['ts'] = sample['ts'][:]

response = predict_daily_uckey(
sample=sample, days=day_list, serving_url=cfg['serving_url'], model_stats=model_stats, columns=columns)
predicted = [response[_] for _ in sorted(response)]

response_1 = predict_daily_uckey(
sample=sample, days=day_list, serving_url=cfg_1['serving_url'], model_stats=model_stats_1, columns=columns)
predicted_1 = [response_1[_] for _ in sorted(response_1)]

#print(predicted)
#print(predicted_1)
for i in range(len(predicted)):
err = abs(predicted[i]-predicted_1[i])/(predicted[i])
errs.append(err)

print(sum(errs)/len(errs))


if __name__ == '__main__':

cfg = {
'log_level': 'warn',
'trainready_table': 'dlpm_111021_no_residency_no_mapping_trainready_test_12212021',
'dist_table': 'dlpm_111021_no_residency_no_mapping_tmp_distribution_test_12212021',
'serving_url': 'http://10.193.217.126:8503/v1/models/dl_test_1221:predict',
'max_calls': 1000,
'model_stat_table': 'dlpm_111021_no_residency_no_mapping_model_stat_test_12212021',
'yesterday': 'WILL BE SET IN PROGRAM'}

cfg_1 = {
'serving_url': 'http://10.193.217.126:8504/v1/models/dl_india:predict',
'model_stat_table': 'dlpm_111021_no_residency_no_mapping_model_stat_india'}

sc = SparkContext.getOrCreate()
hive_context = HiveContext(sc)
sc.setLogLevel(cfg['log_level'])

run(cfg=cfg, cfg_1=cfg_1, hive_context=hive_context)

0 comments on commit 0e97e79

Please sign in to comment.