Skip to content

[SYSTEMDS-3256] Adding include vector parameter in cleaning primitives #1490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Shafaq-Siddiqi wants to merge 4 commits into from
Closed

[SYSTEMDS-3256] Adding include vector parameter in cleaning primitives #1490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bc2bf44
[SYSTEMDS-3256] Adding include vector parameter in cleaning primitives
Shafaq-Siddiqi Dec 16, 2021
1a3d29a
fixing test failures
Shafaq-Siddiqi Dec 22, 2021
0523878
fix test try-2
Shafaq-Siddiqi Dec 22, 2021
e696115
test fix try-03
Shafaq-Siddiqi Dec 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions scripts/builtin/applyAndEvaluate.dml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ s_applyAndEvaluate = function(Frame[Unknown] trainData, Frame[Unknown] testData,
Boolean isLastLabel = TRUE, Boolean correctTypos=FALSE)
return (Matrix[Double] result)
{
no_of_flag_vars = 5
no_of_flag_vars = 6
schema = metaData[1, 1:ncol(metaData) - 1]
mask = as.matrix(metaData[2, 1:ncol(metaData) - 1])
fdMask = as.matrix(metaData[3, 1:ncol(metaData) - 1])
Expand Down Expand Up @@ -144,8 +144,8 @@ return(Double dirtyScore)
mask = as.matrix(metaList['mask'])
mask = ifelse(sum(mask == dmask) < ncol(mask), matrix(1, rows=1, cols=ncol(mask)), mask)
[eXtrain, eXtest] = recodeData(X, Xtest, mask, FALSE, "recode")
eXtrain = replace(target=eXtrain, pattern=NaN, replacement=1)
eXtest = replace(target=eXtest, pattern=NaN, replacement=1)
eXtrain = replace(target=eXtrain, pattern=NaN, replacement=0)
eXtest = replace(target=eXtest, pattern=NaN, replacement=0)
[eXtrain, eXtest] = recodeData(as.frame(eXtrain), as.frame(eXtest), mask, FALSE, "dummycode")
score = eval(evaluationFunc, list(X=eXtrain, Y=Y, Xtest=eXtest, Ytest=Ytest, Xorig=as.matrix(0), evalFunHp=evalFunHp, trainML = FALSE))
dirtyScore = as.scalar(score[1, 1])
Expand Down
32 changes: 17 additions & 15 deletions scripts/builtin/bandit.dml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,16 +268,17 @@ run_with_hyperparam = function(Frame[Unknown] lp, Frame[Unknown] ph_pip, Integer
{
pipList = list(lp = lp, ph = ph_pip[i], hp = hp_matrix, flags = no_of_flag_vars)
[evalFunOutput, hpForPruning, changesByOp] = crossV(X=X, y=Y, cvk=cvk, evalFunHp=evalFunHp, pipList=pipList, metaList=metaList, hpForPruning=hpForPruning,
changesByOp=changesByOp, evalFunc=evaluationFunc, trainML = 0)
changesByOp=changesByOp, evalFunc=evaluationFunc)
}
else
{
[eXtrain, eYtrain, eXtest, eYtest, Tr, hpForPruning, changesByOp] = executePipeline(logical=lp, pipeline=ph_pip[i], X=X, Y=Y, Xtest=Xtest, Ytest=Ytest, metaList=metaList,
hyperParameters=hp_matrix, hpForPruning=hpForPruning, changesByOp=changesByOp, flagsCount=no_of_flag_vars, test=TRUE, verbose=FALSE)
[eXtrain, eYtrain, eXtest, eYtest, Tr, hpForPruning, changesByOp] = executePipeline(logical=lp, pipeline=ph_pip[i],
X=X, Y=Y, Xtest=Xtest, Ytest=Ytest, metaList=metaList, hyperParameters=hp_matrix, hpForPruning=hpForPruning,
changesByOp=changesByOp, flagsCount=no_of_flag_vars, test=TRUE, verbose=FALSE)
if(max(eYtrain) == min(eYtrain))
print("Y contains only one class")
else
evalFunOutput = eval(evaluationFunc, list(X=eXtrain, Y=eYtrain, Xtest=eXtest, Ytest=eYtest, Xorig=as.matrix(0), evalFunHp=evalFunHp, trainML = 0))
evalFunOutput = eval(evaluationFunc, list(X=eXtrain, Y=eYtrain, Xtest=eXtest, Ytest=eYtest, Xorig=as.matrix(0), evalFunHp=evalFunHp))
}

# evalFunOutput = eval(evaluationFunc, argList)
Expand Down Expand Up @@ -320,8 +321,8 @@ getHyperparam = function(Frame[Unknown] pipeline, Frame[Unknown] hpList, Intege
{

allParam = 0;
START_INDEX = 8 # value from where the hyper-params starts after skipping meta flags
NUM_META_FLAGS = 5
START_INDEX = 9 # value from where the hyper-params starts after skipping meta flags
NUM_META_FLAGS = 6
# load the hyper-parameters values
paramList = list()
# store the row indexes of the operator matches
Expand Down Expand Up @@ -354,11 +355,12 @@ getHyperparam = function(Frame[Unknown] pipeline, Frame[Unknown] hpList, Intege
index = as.scalar(indexes[i])
no_of_param = as.integer(as.scalar(paramCount[i]))
# extract hasY and verbose flags
attachMask = matrix(as.scalar(hpList[index, 3]), rows=no_of_res, cols=1)
attachFD = matrix(as.scalar(hpList[index, 4]), rows=no_of_res, cols=1)
attachY = matrix(as.scalar(hpList[index, 5]), rows=no_of_res, cols=1)
isVerbose = matrix(as.scalar(hpList[index, 6]), rows=no_of_res, cols=1)
dataFlag = matrix(as.scalar(hpList[index, 7]), rows=no_of_res, cols=1)
attachIncludeFlag = matrix(as.scalar(hpList[index, 3]), rows=no_of_res, cols=1)
attachMask = matrix(as.scalar(hpList[index, 4]), rows=no_of_res, cols=1)
attachFD = matrix(as.scalar(hpList[index, 5]), rows=no_of_res, cols=1)
attachY = matrix(as.scalar(hpList[index, 6]), rows=no_of_res, cols=1)
isVerbose = matrix(as.scalar(hpList[index, 7]), rows=no_of_res, cols=1)
dataFlag = matrix(as.scalar(hpList[index, 8]), rows=no_of_res, cols=1)

if(no_of_param > 0) {
paramIdx = START_INDEX
Expand Down Expand Up @@ -399,13 +401,13 @@ getHyperparam = function(Frame[Unknown] pipeline, Frame[Unknown] hpList, Intege
typeIdx = typeIdx + 1
}
# hyper-parameter vector contains no. of hp, values of hp, and flag values
OpParam = cbind(matrix(no_of_param, rows=nrow(OpParam), cols=1),OpParam, attachMask,
OpParam = cbind(matrix(no_of_param, rows=nrow(OpParam), cols=1),OpParam, attachIncludeFlag, attachMask,
attachFD, attachY, isVerbose, dataFlag)
}
else {
# no hyper-parameters, so create a dummy matrix of zeros so flags are always aligned
dummy = matrix(0, rows=no_of_res, cols=max(paramCount)+1)
OpParam = cbind(dummy, attachMask, attachFD, attachY)
OpParam = cbind(dummy, attachIncludeFlag, attachMask, attachFD, attachY)
OpParam = cbind(OpParam, isVerbose, dataFlag)
}
paramMatrix[((i-1)*no_of_res)+1:i*no_of_res, 1:ncol(OpParam)] = OpParam
Expand Down Expand Up @@ -598,7 +600,7 @@ return (String s)
}

crossV = function(Matrix[double] X, Matrix[double] y, Integer cvk, Matrix[Double] evalFunHp, List[Unknown] pipList, List[Unknown] metaList,
Matrix[Double] hpForPruning = as.matrix(0), Matrix[Double] changesByOp = as.matrix(0), String evalFunc, Integer trainML = 0)
Matrix[Double] hpForPruning = as.matrix(0), Matrix[Double] changesByOp = as.matrix(0), String evalFunc)
return (Matrix[Double] output, Matrix[Double] hpForPruning, Matrix[Double] changesByOp)
{
accuracyMatrix = matrix(0, cvk, 1)
Expand Down Expand Up @@ -647,7 +649,7 @@ return (Matrix[Double] output, Matrix[Double] hpForPruning, Matrix[Double] chang
changesByOp=changesByOp, flagsCount=as.scalar(pipList['flags']), test=TRUE, verbose=FALSE)
}
# print("test out: "+nrow(testy))
res = eval(evalFunc, list(X=trainX, Y=trainy, Xtest=testX, Ytest=testy, Xorig=as.matrix(0), evalFunHp=evalFunHp, trainML = trainML))
res = eval(evalFunc, list(X=trainX, Y=trainy, Xtest=testX, Ytest=testy, Xorig=as.matrix(0), evalFunHp=evalFunHp))
accuracyMatrix[i] = res[1, 1]
evalFunHp = res[, 2:ncol(res)]
}
Expand Down
150 changes: 132 additions & 18 deletions scripts/builtin/executePipeline.dml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,19 @@ s_executePipeline = function(Frame[String] logical = as.frame("NULL"), Frame[Str
{
mask=as.matrix(metaList['mask'])
FD = as.matrix(metaList['fd'])

includeVector = rbind(matrix(1, rows=nrow(X), cols=1), matrix(0, rows=nrow(Xtest), cols=1))
cloneY = Y
Xorig = X
# # combine X and Y
n = nrow(X)
d = ncol(Xorig)
X = rbind(X, Xtest)

Y = rbind(Y, Ytest)
testRow = nrow(Xtest)
Xout = X
t1 = time()

print("PIPELINE EXECUTION START ... "+toString(pipeline))
if(verbose) {
print("checks rows in X = "+nrow(X)+" rows in Y = "+nrow(Y)+" cols in X = "+ncol(X)+" col in Y = "+ncol(Y))
Expand All @@ -49,9 +51,9 @@ s_executePipeline = function(Frame[String] logical = as.frame("NULL"), Frame[Str
op = as.scalar(pipeline[1,i])
lgOp = as.scalar(logical[1,i])

if(test == FALSE | lgOp != "CI") {
if(lgOp != "CI") {
Xclone = X
[hp, dataFlag, yFlag, executeFlag] = matrixToList(X, Y, mask, FD, hyperParameters[i], flagsCount, op)
[hp, dataFlag, yFlag, executeFlag] = matrixToList(X, Y, includeVector, mask, FD, hyperParameters[i], flagsCount, op)
if(executeFlag == 1) {
X = eval(op, hp)
Xout = X
Expand All @@ -75,7 +77,7 @@ s_executePipeline = function(Frame[String] logical = as.frame("NULL"), Frame[Str
Ytest = Y[testStIdx:nrow(X), ]
X = X[1:trainEndIdx, ]
Y = Y[1:trainEndIdx, ]
[hp, dataFlag, yFlag, executeFlag] = matrixToList(X, Y, mask, FD, hyperParameters[i], flagsCount, op)
[hp, dataFlag, yFlag, executeFlag] = matrixToList(X, Y, includeVector, mask, FD, hyperParameters[i], flagsCount, op)
if(executeFlag == 1)
{
X = eval(op, hp)
Expand All @@ -87,8 +89,10 @@ s_executePipeline = function(Frame[String] logical = as.frame("NULL"), Frame[Str
X = X[, 1:ncol(X) - 1]
}
X = confirmMeta(X, mask)
includeVector = rbind(matrix(1, rows=nrow(X), cols=1), matrix(0, rows=nrow(Xtest), cols=1))
X = rbind(X, Xtest)
Y = rbind(Y, Ytest)

}
else {
print("not applying "+op+" executeFlag = 0")
Expand All @@ -112,7 +116,7 @@ s_executePipeline = function(Frame[String] logical = as.frame("NULL"), Frame[Str
}

# This function will convert the matrix row-vector into list
matrixToList = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] mask, Matrix[Double] FD,
matrixToList = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] includeVector, Matrix[Double] mask, Matrix[Double] FD,
Matrix[Double] p, Integer flagsCount, String op)
return (List[Unknown] l, Integer dataFlag, Integer yFlag, Integer executeFlag)
{
Expand All @@ -122,6 +126,7 @@ matrixToList = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] mask
yFlag = as.integer(as.scalar(p[1, ncol(p) - 2]))
fDFlag = as.integer(as.scalar(p[1, ncol(p)-3]))
maskFlag = as.integer(as.scalar(p[1, ncol(p)-4]))
includeFlag = as.integer(as.scalar(p[1, ncol(p)-5]))

executeFlag = 1
######################################################
Expand Down Expand Up @@ -153,6 +158,14 @@ matrixToList = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] mask
if(yFlag == 1) {
l = append(l, Y)
}

######################################################
# CHECK FOR include APPEND FLAG
if(includeFlag == 1)
{
l = append(l, includeVector)
}

######################################################
# CHECK FOR FD APPEND FLAG
if(fDFlag == 1)
Expand All @@ -166,6 +179,8 @@ matrixToList = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] mask
{
l = append(l, mask)
}


#####################################################
# POPULATE HYPER PARAM
# get the number of hyper-parameters and loop till that
Expand Down Expand Up @@ -270,19 +285,26 @@ return (Matrix[Double] X)
# Output: OHEd matrix X
#######################################################################

dummycoding = function(Matrix[Double] X, Matrix[Double] mask)
return (Matrix[Double] dX_train) {
dummycoding = function(Matrix[Double] X, Matrix[Double] include = as.matrix(1), Matrix[Double] mask)
return (Matrix[Double] output) {


if(sum(mask) > 0)
{
X = replace(target=X, pattern=NaN, replacement=1)
X = replace(target=X, pattern=NaN, replacement=0)

# divide data into train and test
Xtrain = removeEmpty(target=X, margin="rows", select = include)
Xtest = removeEmpty(target=X, margin="rows", select = (include == 0))
idx = vectorToCsv(mask)
# specifications for one-hot encoding of categorical features
jspecDC = "{ids:true, dummycode:["+idx+"]}";
# OHE of categorical features
[dX_train, dM] = transformencode(target=as.frame(X), spec=jspecDC);
[dX_train, dM] = transformencode(target=as.frame(Xtrain), spec=jspecDC);
dX_test = transformapply(target=as.frame(Xtest), spec = jspecDC, meta=dM);
output = rbind(dX_train, dX_test)
}
else dX_train = X
else output = X
}


Expand All @@ -292,21 +314,26 @@ return (Matrix[Double] dX_train) {
# Output: filled matrix X
#######################################################################

imputeByFd = function(Matrix[Double] X, Matrix[Double] fdMask, Double threshold)
imputeByFd = function(Matrix[Double] X, Matrix[Double] include = as.matrix(1), Matrix[Double] fdMask, Double threshold)
return (Matrix[Double] X_filled)
{
if(sum(fdMask) > 0)
{
if(nrow(include) == 1 & ncol(include) == 1)
D = X
else
D = removeEmpty(target=X, margin="rows", select = include)

fdMask = removeEmpty(target=fdMask, margin="cols")
FD = discoverFD(X=replace(target=X, pattern=NaN, replacement=1), Mask=fdMask, threshold=threshold)
FD = discoverFD(X=replace(target=D, pattern=NaN, replacement=1), Mask=fdMask, threshold=threshold)
FD = (diag(matrix(1, rows=nrow(FD), cols=1)) ==0) * FD
FD = FD > 0
if(sum(FD) > 0)
{
for(i in 1: nrow(FD))
{
for(j in 1:ncol(FD)) {
if(as.scalar(FD[i, j]) > 0 & (min(X[, i]) != 0) & (min(X[, j]) != 0) & (sum(FD[, j]) != nrow(FD))
if(as.scalar(FD[i, j]) > 0 & (min(D[, i]) != 0) & (min(D[, j]) != 0) & (sum(FD[, j]) != nrow(FD))
& (as.scalar(fdMask[1, j]) != 0) & (as.scalar(fdMask[1, i]) != 0))
X = imputeByFD(X, i, j, threshold, FALSE)
}
Expand Down Expand Up @@ -404,7 +431,7 @@ m_pca = function(Matrix[Double] X, Integer K=2, Boolean center=TRUE, Boolean sca
D = ncol(X);

# perform z-scoring (centering and scaling)
[X, Centering, ScaleFactor] = scale(X, center, scale);
[X, Centering, ScaleFactor] = scale(X, as.matrix(1), center, scale);

# co-variance matrix
mu = colSums(X)/N;
Expand Down Expand Up @@ -451,11 +478,9 @@ return(Matrix[Double] hpForPruning, Matrix[Double] changesByOp)
########################################################
# The function will flip the noisy labels
########################################################
flipLabels = function(Matrix[Double] X, Matrix[Double] Y, Double threshold, Boolean verbose = FALSE)
flipLabels = function(Matrix[Double] X, Matrix[Double] Y, Double threshold, Integer maxIter =10, Boolean verbose = FALSE)
return (Matrix[Double] XY)
{

print("---- starting flip labels ---")
max_y = max(Y)
if(min(Y) != max(Y))
{
Expand All @@ -464,7 +489,7 @@ return (Matrix[Double] XY)
inc = ((yhat != Y) & (rowMaxs(prob) > threshold))
Xcor = removeEmpty(target = X, margin = "rows", select = (inc==0))
Ycor = removeEmpty(target = Y, margin = "rows", select = (inc==0))
while(sum(inc) > 0)
while(sum(inc) > 0 & maxIter > 0)
{
# print("inc vector "+toString(inc))
Xinc = removeEmpty(target = X, margin = "rows", select = inc)
Expand All @@ -478,9 +503,98 @@ return (Matrix[Double] XY)
Ycor = rbind(Ycor, YcorI)
X = Xinc
Y = Yinc
print("maxIter: "+maxIter)
maxIter = maxIter - 1
}
XY = cbind(Xcor, Ycor)
}
else
XY = cbind(X, Y)
}

#######################################################################
# function frequency conversion
# Inputs: The input dataset X, and mask of the columns
# Output: categorical columns are replaced with their frequencies
#######################################################################

frequencyEncoding = function(Matrix[Double] X, Matrix[Double] include = as.matrix(1), Matrix[Double] mask)
return (Matrix[Double] freqX) {

freqX = X
X = replace(target=X, pattern=NaN, replacement=1)
if(sum(mask) > 0)
{
# get the train data for frequency count
Xtrain = removeEmpty(target=X, margin="rows", select = include)

for(i in 1:ncol(mask))
{
if(as.scalar(mask[1, i]) == 1)
{
Y = X[, i]
valueCount = table(Xtrain[, i], 1)
resp = matrix(0, nrow(Y), max(Y))
resp = (resp + t(seq(1, max(Y)))) == Y
resp = resp * t(valueCount)
freqX[, i] = rowSums(resp)
}

}
}
}

#######################################################################
# function Weight of evidence / information gain
# Inputs: The input dataset X, and mask of the columns
# Output: categorical columns are replaced with their frequencies
#######################################################################

WoE = function(Matrix[Double] X, Matrix[Double] Y, Matrix[Double] include = as.matrix(1), Matrix[Double] mask)
return (Matrix[Double] output) {

freqX = X
X = replace(target=X, pattern=NaN, replacement=1)
if(sum(mask) > 0)
{
# get the train data for frequency count
Xtrain = removeEmpty(target=X, margin="rows", select = include)
Ytrain = removeEmpty(target=Y, margin="rows", select = include)

for(i in 1:ncol(mask))
{
if(as.scalar(mask[1, i]) == 1)
{
L = X[, i]
entropy = getEntropy(Xtrain[, i], Ytrain)
resp = matrix(0, nrow(L), max(L))
resp = (resp + t(seq(1, max(L)))) == L
resp = resp * entropy[, 1:ncol(resp)]
freqX[, i] = rowSums(resp)
}

}
}
output = cbind(freqX, Y)
}


getEntropy = function(Matrix[Double] eX, Matrix[Double] eY)
return(Matrix[Double] entropyMatrix)
{

tab = table(eX, eY)
# print("tab \n"+toString(tab))
entropyMatrix = matrix(0, rows=1, cols=nrow(tab))
catTotal = rowSums(tab)
for(i in 1:nrow(tab))
{
# print("catProb: " +catProb)
entropy = (tab[i,]/catTotal[i])
# print("entropy: "+toString(entropy))
catEntropy = sum(-entropy * log(entropy, 2))
catEntropy = ifelse(is.na(catEntropy), 0, catEntropy)
# print("cat entropy: "+catEntropy)
entropyMatrix[1, i] = catEntropy
}
}
Loading