mirror of
https://github.com/Microsoft/sql-server-samples.git
synced 2025-12-08 14:58:54 +00:00
initial check-in of the mleap_sql sample code.
This commit is contained in:
Lixin Gong
@@ -1,6 +1,14 @@
|
||||
# MLeap on SQL Server Big Data cluster
|
||||
This folder shows how we can build a model with Spark ML and then score the model in SQL Server with its [Java Language Extension](https://docs.microsoft.com/en-us/sql/language-extensions/language-extensions-overview?view=sqlallproducts-allversions)
|
||||
This folder shows how we can build a model with [Spark ML](https://spark.apache.org/docs/latest/ml-guide.html), export the model to [MLeap](https://github.com/combust/mleap), and score the model in SQL Server with its [Java Language Extension](https://docs.microsoft.com/en-us/sql/language-extensions/language-extensions-overview?view=sqlallproducts-allversions)
|
||||
|
||||
## Model training with Spark ML
|
||||
In this sample code, AdultCensusIncome.csv is used to build a Spark ML pipeline model. We can [download the dataset from internet](mleap_sql_test/setup.sh#L11) and [put it on HDFS on a SQL BDC cluster](mleap_sql_test/setup.sh#L12) so that it can be accessed by Spark.
|
||||
|
||||
The data is first [read into Spark](mleap_sql_test/mleap_pyspark.py#L25) and [split into training and testing datasets](mleap_sql_test/mleap_pyspark.py#L64). We then [train a pipeline mode with the training data](mleap_sql_test/mleap_pyspark.py#L87) and [export the model to a mleap bundle](mleap_sql_test/mleap_pyspark.py#L204).
|
||||
|
||||
## Model scoring with SQL Server
|
||||
Now that we have the Spark ML pipeline model in a common serialization [MLeap bundle](http://mleap-docs.combust.ml/core-concepts/mleap-bundles.html) format, we can score the model in Java without the presence of Spark.
|
||||
|
||||
In order to score the model in SQL Server with its [Java Language Extension](https://docs.microsoft.com/en-us/sql/language-extensions/language-extensions-overview?view=sqlallproducts-allversions), we need first build a Java application that can load the model into Java and score it. The [mssql-mleap-app folder](mssql-mleap-app/build.sbt) shows how that can be done.
|
||||
|
||||
Then in T-SQL we can [call the Java application and score the model with some database table](mleap_sql_test/mleap_sql_tests.py#L101).
|
||||
|
||||
Binary file not shown.
BIN
Binary file not shown.
@@ -0,0 +1,6 @@
|
||||
#!/bin/bash
|
||||
|
||||
echo "Cleaning up mleap_sql tests"
|
||||
|
||||
hadoop fs -rm /user/root/AdultCensusIncome.csv
|
||||
rm AdultCensusIncome.csv
|
||||
+237
@@ -0,0 +1,237 @@
|
||||
## train a pyspark model and export it as a mleap bundle
|
||||
import os
|
||||
|
||||
# parse command line arguments
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description = 'train pyspark model and export mleap bundle')
|
||||
parser.add_argument('hdfs_path', nargs='?', default = "/spark_ml", type = str)
|
||||
parser.add_argument('model_name_export', nargs='?', default = "adult_census_pipeline.zip", type = str)
|
||||
args = parser.parse_args()
|
||||
|
||||
hdfs_path = args.hdfs_path
|
||||
model_name_export = args.model_name_export
|
||||
|
||||
# create spark session (needed only if this file is submitted as a spark jobs)
|
||||
from pyspark.sql import SparkSession
|
||||
|
||||
spark = SparkSession\
|
||||
.builder\
|
||||
.appName(os.path.basename(__file__))\
|
||||
.getOrCreate()
|
||||
|
||||
###############################################################################
|
||||
## prepare data
|
||||
|
||||
# read the data into a spark data frame.
|
||||
cwd = os.getcwd()
|
||||
filename = "AdultCensusIncome.csv"
|
||||
|
||||
## NOTE: reading text file from local file path seems flaky!
|
||||
#import urllib.request
|
||||
#url = "https://amldockerdatasets.azureedge.net/" + filename
|
||||
#local_filename, headers = urllib.request.urlretrieve(url, filename)
|
||||
#datafile = "file://" + os.path.join(cwd, filename)
|
||||
|
||||
data_all = spark.read.format('csv')\
|
||||
.options(
|
||||
header='true',
|
||||
inferSchema='true',
|
||||
ignoreLeadingWhiteSpace='true',
|
||||
ignoreTrailingWhiteSpace='true')\
|
||||
.load(filename) #.load(datafile) for local file
|
||||
|
||||
print("Number of rows: {}, Number of coulumns : {}".format(data_all.count(), len(data_all.columns)))
|
||||
|
||||
#replace "-" with "_" in column names
|
||||
columns_new = [col.replace("-", "_") for col in data_all.columns]
|
||||
data_all = data_all.toDF(*columns_new)
|
||||
|
||||
data_all.printSchema()
|
||||
data_all.show(5)
|
||||
|
||||
# choose feature columns and the label column for training.
|
||||
label = "income"
|
||||
#xvars = ["age", "hours_per_week"] #all numeric
|
||||
xvars = ["age", "hours_per_week", "education"] #numeric + string
|
||||
|
||||
print("label: {}, features: {}".format(label, xvars))
|
||||
|
||||
select_cols = xvars
|
||||
select_cols.append(label)
|
||||
data = data_all.select(select_cols)
|
||||
|
||||
###############################################################################
|
||||
## split data into train and test.
|
||||
|
||||
train, test = data.randomSplit([0.75, 0.25], seed=123)
|
||||
|
||||
print("train ({}, {})".format(train.count(), len(train.columns)))
|
||||
print("test ({}, {})".format(test.count(), len(test.columns)))
|
||||
|
||||
train_data_path = os.path.join(hdfs_path, "AdultCensusIncomeTrain")
|
||||
test_data_path = os.path.join(hdfs_path, "AdultCensusIncomeTest")
|
||||
|
||||
# write the train and test data sets to intermediate storage and then read
|
||||
train.write.mode('overwrite').orc(train_data_path)
|
||||
test.write.mode('overwrite').orc(test_data_path)
|
||||
|
||||
print("train and test datasets saved to {} and {}".format(train_data_path, test_data_path))
|
||||
|
||||
train_read = spark.read.orc(train_data_path)
|
||||
test_read = spark.read.orc(test_data_path)
|
||||
|
||||
assert train_read.schema == train.schema and train_read.count() == train.count()
|
||||
assert test_read.schema == test.schema and test_read.count() == test.count()
|
||||
|
||||
###############################################################################
|
||||
## train model
|
||||
|
||||
from pyspark.ml import Pipeline, PipelineModel
|
||||
from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, IndexToString, VectorAssembler
|
||||
from pyspark.ml.classification import LogisticRegression
|
||||
|
||||
# create a new Logistic Regression model, which by default uses "features" and "label" columns for training.
|
||||
reg = 0.1
|
||||
lr = LogisticRegression(regParam=reg)
|
||||
|
||||
# encode string columns
|
||||
dtypes = dict(train.dtypes)
|
||||
dtypes.pop(label)
|
||||
|
||||
si_xvars = []
|
||||
ohe_xvars = []
|
||||
featureCols = []
|
||||
for idx,key in enumerate(dtypes):
|
||||
if dtypes[key] == "string":
|
||||
featureCol = "-".join([key, "encoded"])
|
||||
featureCols.append(featureCol)
|
||||
|
||||
tmpCol = "-".join([key, "tmp"])
|
||||
si_xvars.append(StringIndexer(inputCol=key, outputCol=tmpCol, handleInvalid="skip")) #, handleInvalid="keep"
|
||||
ohe_xvars.append(OneHotEncoderEstimator(inputCols=[tmpCol], outputCols=[featureCol]))
|
||||
else:
|
||||
featureCols.append(key)
|
||||
|
||||
# string-index the label column into a column named "label"
|
||||
si_label = StringIndexer(inputCol=label, outputCol='label')
|
||||
#si_label._resetUid("si_label") # try to name the transformer, which seems not carried over to the fitted pipeline.
|
||||
|
||||
# assemble the encoded feature columns in to a column named "features"
|
||||
assembler = VectorAssembler(inputCols=featureCols, outputCol="features")
|
||||
|
||||
# put together the pipeline
|
||||
stages = []
|
||||
stages.extend(si_xvars)
|
||||
stages.extend(ohe_xvars)
|
||||
stages.append(si_label)
|
||||
stages.append(assembler)
|
||||
stages.append(lr)
|
||||
|
||||
pipe = Pipeline(stages=stages)
|
||||
print("Pipeline Created")
|
||||
|
||||
# train the model
|
||||
model = pipe.fit(train)
|
||||
print("Model Trained")
|
||||
print("Model is ", model)
|
||||
print("Model Stages", model.stages)
|
||||
|
||||
# name the string-index stage for the label so it can be identified easier later
|
||||
model.stages[2]._resetUid("si_label")
|
||||
|
||||
###############################################################################
|
||||
## evaluate model
|
||||
|
||||
from pyspark.ml.evaluation import BinaryClassificationEvaluator
|
||||
|
||||
# make prediction
|
||||
pred = model.transform(test)
|
||||
|
||||
# evaluate. note only 2 metrics are supported out of the box by Spark ML.
|
||||
bce = BinaryClassificationEvaluator(rawPredictionCol='rawPrediction')
|
||||
au_roc = bce.setMetricName('areaUnderROC').evaluate(pred)
|
||||
au_prc = bce.setMetricName('areaUnderPR').evaluate(pred)
|
||||
|
||||
print("Area under ROC: {}".format(au_roc))
|
||||
print("Area Under PR: {}".format(au_prc))
|
||||
|
||||
###############################################################################
|
||||
## save and load the model with ML persistence
|
||||
# https://spark.apache.org/docs/latest/ml-pipeline.html#ml-persistence-saving-and-loading-pipelines
|
||||
|
||||
##NOTE: by default the model is saved to and loaded from hdfs
|
||||
model_name = "AdultCensus.mml"
|
||||
model_fs = os.path.join(hdfs_path, model_name)
|
||||
|
||||
model.write().overwrite().save(model_fs)
|
||||
print("saved model to {}".format(model_fs))
|
||||
|
||||
# load the model file (from hdfs)
|
||||
print("load pyspark model from hdfs")
|
||||
model_loaded = PipelineModel.load(model_fs)
|
||||
assert str(model_loaded) == str(model)
|
||||
|
||||
print("loaded model from {}".format(model_fs))
|
||||
print("Model is " , model_loaded)
|
||||
print("Model stages", model_loaded.stages)
|
||||
|
||||
###############################################################################
|
||||
## export and import model with mleap
|
||||
|
||||
import mleap.pyspark
|
||||
from mleap.pyspark.spark_support import SimpleSparkSerializer
|
||||
|
||||
# serialize the model to a local zip file in JSON format
|
||||
#model_name_export = "adult_census_pipeline.zip"
|
||||
model_name_path = cwd
|
||||
model_file = os.path.join(model_name_path, model_name_export)
|
||||
|
||||
# remove an old model file, if needed.
|
||||
if os.path.isfile(model_file):
|
||||
os.remove(model_file)
|
||||
|
||||
model_file_path = "jar:file:{}".format(model_file)
|
||||
model.serializeToBundle(model_file_path, model.transform(train))
|
||||
|
||||
## import mleap model
|
||||
model_deserialized = PipelineModel.deserializeFromBundle(model_file_path)
|
||||
assert str(model_deserialized) == str(model)
|
||||
|
||||
print("The deserialized model is ", model_deserialized)
|
||||
print("The deserialized model stages are", model_deserialized.stages)
|
||||
|
||||
##############################################################################
|
||||
## export the final model with mleap
|
||||
|
||||
## remove the stringIndexer for the label column so it won't be required for prediction
|
||||
model_final = model.copy()
|
||||
|
||||
si_label_index = -3
|
||||
model_final.stages.pop(si_label_index) #si_label
|
||||
|
||||
## append an IndexToString transformer to the model pipeline to get the original labels
|
||||
#labelReverse = IndexToString(inputCol = "label", outputCol = "predIncome") #no need to provide labels
|
||||
labelReverse = IndexToString(
|
||||
inputCol = "prediction",
|
||||
outputCol = "predictedIncome",
|
||||
labels = model.stages[si_label_index].labels) #must provide labels (from si_label) otherwise will fail
|
||||
model_final.stages.append(labelReverse)
|
||||
|
||||
pred_final = model_final.transform(test)
|
||||
pred_final.printSchema()
|
||||
pred_final.show(5)
|
||||
|
||||
# remove an old model file, if needed.
|
||||
if os.path.isfile(model_file):
|
||||
os.remove(model_file)
|
||||
model_final.serializeToBundle(model_file_path, model_final.transform(train))
|
||||
|
||||
print("persist the mleap bundle from local to hdfs")
|
||||
from subprocess import Popen, PIPE
|
||||
hdfs_fs_put = ["hadoop", "fs", "-put", "-f", model_file, os.path.join(hdfs_path, model_name_export)]
|
||||
proc = Popen(hdfs_fs_put, stdout=PIPE, stderr=PIPE)
|
||||
s_output, s_err = proc.communicate()
|
||||
if (s_err):
|
||||
print("s_output: {s_output}\ns_err: {s_err}".format(s_output=s_output, s_err=s_err))
|
||||
|
||||
###############################################################################
|
||||
+164
@@ -0,0 +1,164 @@
|
||||
import os
|
||||
dir_path = os.path.dirname(os.path.realpath(__file__))
|
||||
|
||||
import sys
|
||||
sys.path.append(os.path.join(dir_path, os.pardir, os.pardir, os.pardir))
|
||||
|
||||
from spark_submit import *
|
||||
|
||||
from subprocess import run, PIPE
|
||||
import pytest
|
||||
import pyodbc
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def setup_mod():
|
||||
print("setting up module ...")
|
||||
|
||||
odbcDriver = "ODBC Driver 13 for SQL Server"
|
||||
databaseName = "tempdb"
|
||||
headNode = "master-0.master-svc"
|
||||
|
||||
# Read sql username and password from environment variable.
|
||||
username = os.environ["EXTENSIBILITY_TEST_SQL_USER"]
|
||||
password = os.environ["EXTENSIBILITY_TEST_SQL_PASSWORD"]
|
||||
if not username or not password:
|
||||
raise Exception("Environment variable EXTENSIBILITY_TEST_SQL_USER or EXTENSIBILITY_TEST_SQL_PASSWORD cannot not be found")
|
||||
|
||||
# enable SPEES
|
||||
conn = pyodbc.connect('DRIVER={0};SERVER={1};DATABASE={2};UID={3};PWD={4}'.format(
|
||||
odbcDriver, headNode, databaseName, username, password), autocommit=True)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""EXEC sp_configure 'external scripts enabled', 1""")
|
||||
assert(-1 == cursor.rowcount)
|
||||
|
||||
cursor.execute("""RECONFIGURE""")
|
||||
assert(-1 == cursor.rowcount)
|
||||
|
||||
yield dict(cursor=cursor)
|
||||
|
||||
print("tearing down module ...")
|
||||
|
||||
|
||||
def test_java_spees(setup_mod):
|
||||
# exectue a Java SPEES query to create external libraries
|
||||
cursor = setup_mod['cursor']
|
||||
cursor.execute("""
|
||||
--SELECT @@SERVERNAME AS 'Server Name', @@VERSION AS 'Server Version', @@SERVICENAME AS 'Service Name'
|
||||
|
||||
IF NOT EXISTS (SELECT * FROM sys.external_languages WHERE language = 'Java')
|
||||
--DROP EXTERNAL LANGUAGE Java;
|
||||
CREATE EXTERNAL LANGUAGE Java
|
||||
FROM (CONTENT = N'/opt/mssql/lib/extensibility/java-lang-extension.tar.gz', file_name = 'javaextension.so');
|
||||
|
||||
IF EXISTS (SELECT * FROM sys.external_libraries WHERE name = 'SdkPackage')
|
||||
DROP EXTERNAL LIBRARY SdkPackage;
|
||||
CREATE EXTERNAL LIBRARY SdkPackage
|
||||
FROM (CONTENT = '/opt/mssql/lib/mssql-java-lang-extension.jar') WITH (LANGUAGE = 'Java');
|
||||
|
||||
IF EXISTS (SELECT * FROM sys.external_libraries WHERE name = 'TestPackage')
|
||||
DROP EXTERNAL LIBRARY TestPackage
|
||||
CREATE EXTERNAL LIBRARY TestPackage
|
||||
FROM (CONTENT = '/opt/mssql/java/jars/JavaTestPackage.jar') WITH (LANGUAGE = 'Java');
|
||||
|
||||
DECLARE @script NVARCHAR(max) = N'JavaTestPackage.PassThrough' --no space allowed in the string!
|
||||
EXEC sp_execute_external_script
|
||||
@language = N'Java'
|
||||
, @script = @script
|
||||
, @input_data_1 = N'SELECT 1'
|
||||
""")
|
||||
|
||||
rows = cursor.fetchall()
|
||||
assert(1 == len(rows))
|
||||
assert(1 == rows[0][0])
|
||||
|
||||
|
||||
def dictfetchall(cursor):
|
||||
'''fetch all rows from a cursor and return them as a dict'''
|
||||
colnames = [col[0] for col in cursor.description]
|
||||
return [dict(zip(colnames, row)) for row in cursor.fetchall()]
|
||||
|
||||
|
||||
def test_mleap_pyspark(setup_mod):
|
||||
# train a pyspark model and export it as a mleap bundle
|
||||
hdfs_path = "/spark_ml"
|
||||
model_name_export = "adult_census_pipeline.zip"
|
||||
|
||||
file_path = 'mleap_pyspark.py'
|
||||
file_args = [hdfs_path, model_name_export]
|
||||
ret = spark_submit(file_path, file_args)
|
||||
assert 0 == ret
|
||||
|
||||
# get the mleap bundle from hdfs and copy it to the mssql-server container of the master-0 pod
|
||||
hdfs_file_path = os.path.join(hdfs_path, model_name_export)
|
||||
ret = run(["hdfs", "dfs", "-get", "-f", hdfs_file_path], stdout=PIPE, stderr=PIPE).returncode
|
||||
assert 0 == ret
|
||||
|
||||
local_file_path = os.path.join("master-0:", "tmp")
|
||||
ret = run(["kubectl", "cp", model_name_export, local_file_path, "-c", "mssql-server"], stdout=PIPE, stderr=PIPE).returncode
|
||||
assert 0 == ret
|
||||
|
||||
# exectue a Java SPEES query to serve the mleap bundle
|
||||
cursor = setup_mod['cursor']
|
||||
cursor.execute("""
|
||||
--suppresses the record count values generated by DML statements
|
||||
--like UPDATE and allows the result set to be retrieved directly.
|
||||
SET NOCOUNT ON;
|
||||
|
||||
IF EXISTS (SELECT * FROM sys.external_libraries WHERE name = 'MleapApp')
|
||||
DROP EXTERNAL LIBRARY MleapApp;
|
||||
CREATE EXTERNAL LIBRARY MleapApp
|
||||
FROM (CONTENT = '/opt/mssql/java/jars/mssql-mleap-app-assembly-1.0.jar') WITH (LANGUAGE = 'Java')
|
||||
|
||||
DROP TABLE IF EXISTS ##test
|
||||
CREATE TABLE ##test (
|
||||
income nvarchar(10)
|
||||
, age int
|
||||
, hours_per_week int
|
||||
, education nvarchar(10)
|
||||
, sex nvarchar(10)
|
||||
);
|
||||
INSERT INTO ##test values ('<=50K', 39, 40, 'Bachelors', 'Male');
|
||||
INSERT INTO ##test values ('<=50K', 50, 13, 'Bachelors', 'Male');
|
||||
INSERT INTO ##test values ('<=50K', 38, 40, 'HS-grad', 'Male');
|
||||
--SELECT * FROM ##test
|
||||
|
||||
DECLARE @script NVARCHAR(max) = N'com.microsoft.sqlserver.mleap.Scorer' --no space allowed in the string!
|
||||
DECLARE @language nvarchar(4) = N'Java'
|
||||
DECLARE @parallel bit = 0
|
||||
DECLARE @input_data_1 nvarchar(97) = N'select age, hours_per_week, education, sex, income from ##test'
|
||||
DECLARE @params nvarchar(200) = N'@modelPath nvarchar(100), @outputFields nvarchar(100), @logLevel nvarchar(100)'
|
||||
DECLARE @modelPath nvarchar(100) = N'/tmp/adult_census_pipeline.zip'
|
||||
DECLARE @outputFields nvarchar(100) = N'prediction,probability,education,sex,income,predictedIncome'
|
||||
DECLARE @logLevel nvarchar(100) = N'INFO'
|
||||
EXEC sp_execute_external_script @language = @language, @script = @script, @parallel = @parallel
|
||||
, @input_data_1 = @input_data_1
|
||||
, @params = @params, @modelPath = @modelPath, @outputFields = @outputFields, @logLevel = @logLevel
|
||||
WITH RESULT SETS ((prediction int, probability0 float, probability1 float, education nvarchar(20), sex nvarchar(20), income nvarchar(20), predictedIncome nvarchar(20)))
|
||||
""")
|
||||
|
||||
rows = dictfetchall(cursor)
|
||||
#pandas.DataFrame(rows)
|
||||
|
||||
assert rows == [
|
||||
{'education': 'Bachelors',
|
||||
'income': '<=50K',
|
||||
'predictedIncome': '<=50K',
|
||||
'prediction': 0,
|
||||
'probability0': 0.6544871023375456,
|
||||
'probability1': 0.3455128976624544,
|
||||
'sex': 'Male'},
|
||||
{'education': 'Bachelors',
|
||||
'income': '<=50K',
|
||||
'predictedIncome': '<=50K',
|
||||
'prediction': 0,
|
||||
'probability0': 0.7363751868447964,
|
||||
'probability1': 0.2636248131552036,
|
||||
'sex': 'Male'},
|
||||
{'education': 'HS-grad',
|
||||
'income': '<=50K',
|
||||
'predictedIncome': '<=50K',
|
||||
'prediction': 0,
|
||||
'probability0': 0.8324466132959966,
|
||||
'probability1': 0.16755338670400344,
|
||||
'sex': 'Male'}]
|
||||
@@ -0,0 +1,17 @@
|
||||
#!/bin/bash -e
|
||||
|
||||
echo "Setting up mleap_sql tests"
|
||||
|
||||
export PYSPARK_PYTHON=python3
|
||||
|
||||
export EXTENSIBILITY_TEST_SQL_USER=sa
|
||||
export EXTENSIBILITY_TEST_SQL_PASSWORD=Yukon900
|
||||
|
||||
hadoop fs -mkdir -p /user/root
|
||||
wget https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv
|
||||
hadoop fs -copyFromLocal AdultCensusIncome.csv /user/root
|
||||
|
||||
# Copy java ext jars to mssql-server container in master pod
|
||||
#kubectl cp -c mssql-server ../jars/mssql_java_lang_extension.jar master-0:/opt/mssql/java/jars/
|
||||
kubectl cp -c mssql-server ../jars/JavaTestPackage.jar master-0:/opt/mssql/java/jars/
|
||||
kubectl cp -c mssql-server ../mssql-mleap-app/target/scala-2.11/mssql-mleap-app-assembly-1.0.jar master-0:/opt/mssql/java/jars/
|
||||
@@ -0,0 +1,6 @@
|
||||
#!/bin/bash
|
||||
|
||||
source ./setup.sh
|
||||
|
||||
# Generate Junit results
|
||||
python3 -m pytest -v --junitxml /tests/junit/mleap_sql.xml -o junit_suite_name=mleap_sql --durations=0 mleap_sql_tests.py
|
||||
@@ -0,0 +1,16 @@
|
||||
.DEFAULT_GOAL = all
|
||||
|
||||
all: assembly
|
||||
|
||||
assembly:
|
||||
@sbt assembly
|
||||
|
||||
package:
|
||||
@sbt package
|
||||
|
||||
clean:
|
||||
@rm -rf project/project
|
||||
@rm -rf project/target
|
||||
@rm -rf target
|
||||
@rm -rf .idea
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
name := "mssql-mleap-app"
|
||||
|
||||
version := "1.0"
|
||||
|
||||
scalaVersion := "2.11.12"
|
||||
|
||||
libraryDependencies ++= Seq(
|
||||
"ml.combust.mleap" %% "mleap-runtime" % "0.13.0" % "provided",
|
||||
"org.apache.commons" % "commons-csv" % "1.5",
|
||||
"commons-cli" % "commons-cli" % "1.4",
|
||||
"org.scalatest" %% "scalatest" % "3.2.0-SNAP10" % Test,
|
||||
"org.scalacheck" %% "scalacheck" % "1.14.0" % Test,
|
||||
"com.novocode" % "junit-interface" % "0.11" % Test
|
||||
)
|
||||
|
||||
// Exclude scala-library from this fat jar. The scala library is already there in spark package.
|
||||
assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false)
|
||||
|
||||
// exclude specific jars
|
||||
assemblyExcludedJars in assembly := {
|
||||
val cp = (fullClasspath in assembly).value
|
||||
cp filter {_.data.getName == "mssql_java_lang_extension.jar"}
|
||||
}
|
||||
BIN
Binary file not shown.
+1
@@ -0,0 +1 @@
|
||||
sbt.version = 1.1.5
|
||||
+1
@@ -0,0 +1 @@
|
||||
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.6")
|
||||
+85
@@ -0,0 +1,85 @@
|
||||
package com.microsoft.sqlserver.mleap;
|
||||
|
||||
import java.sql.JDBCType;
|
||||
import java.sql.Types;
|
||||
import java.util.Arrays;
|
||||
|
||||
public class PrimitiveDataset extends com.microsoft.sqlserver.javalangextension.PrimitiveDataset {
|
||||
public String[] getColumnNames() {
|
||||
int nCols = getColumnCount();
|
||||
String[] columnNames = new String[nCols];
|
||||
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
columnNames[iCol] = getColumnName(iCol);
|
||||
}
|
||||
|
||||
return columnNames;
|
||||
}
|
||||
|
||||
public int[] getColumnTypes() {
|
||||
int nCols = getColumnCount();
|
||||
int[] columnTypes = new int[nCols];
|
||||
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
columnTypes[iCol] = getColumnType(iCol);
|
||||
}
|
||||
|
||||
return columnTypes;
|
||||
}
|
||||
|
||||
public int getColumnIndex(String columnName) {
|
||||
String[] columnNames = getColumnNames();
|
||||
int index = Arrays.asList(columnNames).indexOf(columnName);
|
||||
return index;
|
||||
}
|
||||
|
||||
public int getRowCount(int iCol) {
|
||||
int sqlType = getColumnType(iCol);
|
||||
int columnLength;
|
||||
|
||||
switch(sqlType) {
|
||||
case Types.BIT:
|
||||
columnLength = getBooleanColumn(iCol).length;
|
||||
break;
|
||||
case Types.SMALLINT:
|
||||
columnLength = getShortColumn(iCol).length;
|
||||
break;
|
||||
case Types.INTEGER:
|
||||
columnLength = getIntColumn(iCol).length;
|
||||
break;
|
||||
case Types.BIGINT:
|
||||
columnLength = getLongColumn(iCol).length;
|
||||
break;
|
||||
case Types.FLOAT:
|
||||
columnLength = getFloatColumn(iCol).length;
|
||||
break;
|
||||
case Types.DOUBLE:
|
||||
columnLength = getDoubleColumn(iCol).length;
|
||||
break;
|
||||
case Types.NVARCHAR:
|
||||
columnLength = getStringColumn(iCol).length;
|
||||
break;
|
||||
case Types.VARBINARY:
|
||||
columnLength = getBinaryColumn(iCol).length;
|
||||
break;
|
||||
case Types.DATE:
|
||||
columnLength = getDateColumn(iCol).length;
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("unsupported sql type: " + JDBCType.valueOf(sqlType).getName());
|
||||
}
|
||||
|
||||
return columnLength;
|
||||
}
|
||||
|
||||
public int[] getRowCounts() {
|
||||
int nCols = getColumnCount();
|
||||
int[] rowCounts = new int[nCols];
|
||||
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
rowCounts[iCol] = getRowCount(iCol);
|
||||
}
|
||||
|
||||
return rowCounts;
|
||||
}
|
||||
}
|
||||
+314
@@ -0,0 +1,314 @@
|
||||
package com.microsoft.sqlserver.mleap;
|
||||
|
||||
import com.microsoft.sqlserver.javalangextension.AbstractSqlServerExtensionExecutor;
|
||||
|
||||
import org.apache.commons.csv.CSVFormat;
|
||||
import org.apache.commons.csv.CSVRecord;
|
||||
import org.apache.commons.cli.*;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.FileReader;
|
||||
import java.io.Reader;
|
||||
|
||||
import java.sql.JDBCType;
|
||||
import java.sql.Types;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class Scorer extends AbstractSqlServerExtensionExecutor {
|
||||
|
||||
private static final Logger LOGGER = Logger.getLogger(Scorer.class.getName());
|
||||
|
||||
public Scorer() {
|
||||
executorExtensionVersion = SQLSERVER_JAVA_LANG_EXTENSION_V1;
|
||||
executorInputDatasetClassName = PrimitiveDataset.class.getName();
|
||||
executorOutputDatasetClassName = PrimitiveDataset.class.getName();
|
||||
}
|
||||
|
||||
public void init(String sessionId, int taskId, int numTasks) {
|
||||
System.out.println("init SessionID: " + sessionId + " taskId: " + taskId + " numTasks: " + numTasks);
|
||||
}
|
||||
|
||||
public PrimitiveDataset execute(PrimitiveDataset input, LinkedHashMap<String, Object> params) {
|
||||
List<String> logLevels = Arrays.asList("OFF", "SEVERE", "WARNING", "INFO", "CONFIG", "FINE", "FINER", "FINEST", "ALL");
|
||||
String logLevel = params.getOrDefault("logLevel", "WARNING").toString();
|
||||
if (!logLevels.contains(logLevel)) {
|
||||
throw new IllegalArgumentException("logLevel (" + logLevel + ") must be one of " + logLevels.toString());
|
||||
}
|
||||
LOGGER.setLevel(Level.parse(logLevel));
|
||||
|
||||
LOGGER.info("Logger Name: " + LOGGER.getName() + "; Logger Level:" + LOGGER.getLevel());
|
||||
|
||||
// load model
|
||||
String modelPath;
|
||||
try {
|
||||
modelPath = params.get("modelPath").toString();
|
||||
} catch (NullPointerException e) {
|
||||
throw new IllegalArgumentException("modelPath parameter is required but not set.");
|
||||
}
|
||||
|
||||
long startTime = System.nanoTime();
|
||||
Predictor scorer = new Predictor();
|
||||
scorer.init(modelPath);
|
||||
long endTime = System.nanoTime();
|
||||
long duration = (endTime - startTime); //divide by 10^6 to get milliseconds.
|
||||
LOGGER.info("model loading time: " + duration/1e6 + " ms");
|
||||
|
||||
// convert PrimitiveDataset to DefaultLeapFrame
|
||||
startTime = System.nanoTime();
|
||||
scorer.primitiveDataset2leapFrame(input);
|
||||
endTime = System.nanoTime();
|
||||
duration = (endTime - startTime); //divide by 10^6 to get milliseconds.
|
||||
LOGGER.info("PrimitiveDataset to DefaultLeapFrame conversion time: " + duration/1e6 + " ms");
|
||||
|
||||
// do prediction
|
||||
startTime = System.nanoTime();
|
||||
scorer.run();
|
||||
endTime = System.nanoTime();
|
||||
duration = (endTime - startTime); //divide by 10^6 to get milliseconds.
|
||||
LOGGER.info("model scoring time: " + duration/1e6 + " ms");
|
||||
|
||||
//select output fields specified
|
||||
startTime = System.nanoTime();
|
||||
String[] outputFields;
|
||||
outputFields = params.getOrDefault("outputFields", "").toString().split(",");
|
||||
scorer.select(outputFields);
|
||||
endTime = System.nanoTime();
|
||||
duration = (endTime - startTime); //divide by 10^6 to get milliseconds.
|
||||
LOGGER.info("data selection time: " + duration/1e6 + " ms");
|
||||
|
||||
// convert DefaultLeapFrame to PrimitiveDataset
|
||||
startTime = System.nanoTime();
|
||||
PrimitiveDataset output = new PrimitiveDataset();
|
||||
scorer.leapFrame2primitiveDataset(output);
|
||||
endTime = System.nanoTime();
|
||||
duration = (endTime - startTime); //divide by 10^6 to get milliseconds.
|
||||
LOGGER.info("DefaultLeapFrame to PrimitiveDataset conversion time: " + duration/1e6 + " ms");
|
||||
|
||||
return output;
|
||||
}
|
||||
|
||||
public void cleanup() {
|
||||
System.out.println("\n* cleanup");
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param args commandline options for model path and input file.
|
||||
* @throws Exception
|
||||
* <pre>
|
||||
* {@code
|
||||
*
|
||||
* -- ex: Linux
|
||||
* java -cp mssql-mleap-app-assembly-1.0.jar:mssql_java_lang_extension.jar:mssql-mleap-lib-assembly-1.0.jar:commons-csv-1.5.jar:commons-cli-1.4.jar com.microsoft.sqlserver.mleap.Scorer
|
||||
* -m /tmp/adult_census_pipeline.zip
|
||||
* -i /tmp/adult_census_income.csv
|
||||
*
|
||||
* java -cp "*" -m /tmp/adult_census_pipeline.zip -i /tmp/adult_census_income.csv
|
||||
*
|
||||
* -- ex: Windows
|
||||
* java -cp mssql-mleap-app-assembly-1.0.jar;mssql_java_lang_extension.jar;mssql-mleap-lib-assembly-1.0.jar;commons-csv-1.5.jar:commons-cli-1.4.jar com.microsoft.sqlserver.mleap.Scorer
|
||||
* -m C:\\Users\\lgong\\Work\\git\\aml-databricks\\examples\\mleapsql2\\src\\main\\resources\\sqlqueries\\adult_census_pipeline.zip
|
||||
* -i C:\\Users\\lgong\\Work\\git\\aml-databricks\\examples\\mleapsql2\\src\\main\\resources\\sqlqueries\\adult_census_income.csv
|
||||
*
|
||||
* java -cp "*"
|
||||
* -m C:\\Users\\lgong\\Work\\git\\aml-databricks\\examples\\mleapsql2\\src\\main\\resources\\sqlqueries\\adult_census_pipeline.zip
|
||||
* -i C:\\Users\\lgong\\Work\\git\\aml-databricks\\examples\\mleapsql2\\src\\main\\resources\\sqlqueries\\adult_census_income.csv
|
||||
* }
|
||||
* </pre>
|
||||
*/
|
||||
public static void main(String[] args) throws Exception {
|
||||
// get model and testing data
|
||||
Options options = new Options();
|
||||
|
||||
Option input = new Option("i", "input", true, "input file");
|
||||
input.setRequired(true);
|
||||
options.addOption(input);
|
||||
|
||||
Option model = new Option("m", "model", true, "model path");
|
||||
model.setRequired(true);
|
||||
options.addOption(model);
|
||||
|
||||
CommandLineParser parser = new DefaultParser();
|
||||
HelpFormatter formatter = new HelpFormatter();
|
||||
CommandLine cmd = null;
|
||||
|
||||
try {
|
||||
cmd = parser.parse(options, args);
|
||||
} catch (ParseException e) {
|
||||
System.out.println(e.getMessage());
|
||||
formatter.printHelp("Scorer", options);
|
||||
|
||||
System.exit(1);
|
||||
}
|
||||
|
||||
String modelPath = cmd.getOptionValue("model");
|
||||
String scoreFile = cmd.getOptionValue("input");
|
||||
|
||||
LOGGER.info("os.name: " + System.getProperty("os.name"));
|
||||
LOGGER.info("isWindows: " + System.getProperty("os.name").startsWith("Windows"));
|
||||
LOGGER.info("args: " + Arrays.toString(args));
|
||||
|
||||
LOGGER.info("modelPath: " + modelPath);
|
||||
LOGGER.info("scoreFile: " + scoreFile);
|
||||
|
||||
// read in the testing data
|
||||
BufferedReader bufferedReader = new BufferedReader(new FileReader(scoreFile));
|
||||
int nRows = -1; //account for the header row
|
||||
while(bufferedReader.readLine() != null) {
|
||||
nRows++;
|
||||
}
|
||||
|
||||
LinkedHashMap<String, Integer> inputFields = new LinkedHashMap<String, Integer>();
|
||||
inputFields.put("age", Types.INTEGER);
|
||||
inputFields.put("workclass", Types.NVARCHAR);
|
||||
inputFields.put("fnlwgt", Types.INTEGER);
|
||||
inputFields.put("education", Types.NVARCHAR);
|
||||
inputFields.put("education_num", Types.INTEGER);
|
||||
inputFields.put("marital_status", Types.NVARCHAR);
|
||||
inputFields.put("occupation", Types.NVARCHAR);
|
||||
inputFields.put("relationship", Types.NVARCHAR);
|
||||
inputFields.put("race", Types.NVARCHAR);
|
||||
inputFields.put("sex", Types.NVARCHAR);
|
||||
inputFields.put("capital_gain", Types.INTEGER);
|
||||
inputFields.put("capital_loss", Types.INTEGER);
|
||||
inputFields.put("hours_per_week", Types.INTEGER);
|
||||
inputFields.put("native_country", Types.NVARCHAR);
|
||||
inputFields.put("income", Types.NVARCHAR);
|
||||
|
||||
String[] columnNames = {"age", "hours_per_week", "education", "sex", "income"}; //choose the input variables
|
||||
int[] columnTypes = new int[columnNames.length];
|
||||
for (int iCol = 0; iCol < columnNames.length; iCol++) {
|
||||
try {
|
||||
columnTypes[iCol] = inputFields.get(columnNames[iCol]);
|
||||
} catch (NullPointerException e) {
|
||||
throw new IllegalArgumentException("invalid input field: " + columnNames[iCol]);
|
||||
}
|
||||
}
|
||||
int nCols = columnNames.length;
|
||||
|
||||
Object[] columns = new Object[nCols];
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
int columnType = columnTypes[iCol];
|
||||
switch (columnType) {
|
||||
case Types.INTEGER:
|
||||
columns[iCol] = new int[nRows];
|
||||
break;
|
||||
case Types.NVARCHAR:
|
||||
columns[iCol] = new String[nRows];
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("unsupported sql type: " + JDBCType.valueOf(columnType).getName());
|
||||
}
|
||||
}
|
||||
|
||||
Reader in = new FileReader(scoreFile);
|
||||
Iterable<CSVRecord> records = CSVFormat.RFC4180.withFirstRecordAsHeader().parse(in);
|
||||
int iRow = 0;
|
||||
for (CSVRecord record : records) {
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
int columnType = columnTypes[iCol];
|
||||
switch (columnType) {
|
||||
case Types.INTEGER:
|
||||
((int[])(columns[iCol]))[iRow] = Integer.parseInt(record.get(columnNames[iCol]));
|
||||
break;
|
||||
case Types.NVARCHAR:
|
||||
((String[])(columns[iCol]))[iRow] = record.get(columnNames[iCol]);
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("unsupported sql type: " + JDBCType.valueOf(columnType).getName());
|
||||
}
|
||||
}
|
||||
iRow++;
|
||||
}
|
||||
|
||||
// form the primitive dataset
|
||||
PrimitiveDataset inputds = new PrimitiveDataset();
|
||||
|
||||
for (int iCol = 0; iCol < nCols; iCol++) {
|
||||
int columnType = columnTypes[iCol];
|
||||
switch (columnType) {
|
||||
case Types.INTEGER:
|
||||
inputds.addColumnMetadata(iCol, columnNames[iCol], Types.INTEGER, 0, 0);
|
||||
inputds.addIntColumn(iCol, (int[])(columns[iCol]), null);
|
||||
break;
|
||||
case Types.NVARCHAR:
|
||||
inputds.addColumnMetadata(iCol, columnNames[iCol], Types.NVARCHAR, 0, 0);
|
||||
inputds.addStringColumn(iCol, (String[])(columns[iCol]));
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("unsupported sql type: " + JDBCType.valueOf(columnType).getName());
|
||||
}
|
||||
}
|
||||
|
||||
// specify some params
|
||||
LinkedHashMap<String, Object> params = new LinkedHashMap<>();
|
||||
params.put("logLevel", "INFO"); //default WARN
|
||||
params.put("modelPath", modelPath);
|
||||
params.put("outputFields", "prediction,probability,education,sex,income,predictedIncome");
|
||||
//params.put("outputFields", "features,education-encoded"); //SparseTensor
|
||||
|
||||
// perform scoring
|
||||
Scorer scorer = new Scorer();
|
||||
scorer.init("session0", 0, 1);
|
||||
PrimitiveDataset output = scorer.execute(inputds, params);
|
||||
|
||||
// display output
|
||||
int nOutputCols = output.getColumnCount();
|
||||
System.out.println("\nnOutputCols: " + nOutputCols);
|
||||
|
||||
for (int iCol = 0; iCol < nOutputCols; iCol++) {
|
||||
System.out.println("\nColumnName: " + output.getColumnName(iCol));
|
||||
|
||||
int columnType = output.getColumnType(iCol);
|
||||
System.out.println("ColumnType: " + JDBCType.valueOf(columnType).getName());
|
||||
|
||||
switch(columnType) {
|
||||
case Types.INTEGER:
|
||||
int[] intColumn = output.getIntColumn(iCol);
|
||||
System.out.println("Column.length: " + intColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(intColumn));
|
||||
break;
|
||||
case Types.DOUBLE:
|
||||
double[] doubleColumn = output.getDoubleColumn(iCol);
|
||||
System.out.println("Column.length: " + doubleColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(doubleColumn));
|
||||
break;
|
||||
case Types.BIGINT:
|
||||
long[] longColumn = output.getLongColumn(iCol);
|
||||
System.out.println("Column.length: " + longColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(longColumn));
|
||||
break;
|
||||
case Types.BIT:
|
||||
boolean[] booleanColumn = output.getBooleanColumn(iCol);
|
||||
System.out.println("Column.length: " + booleanColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(booleanColumn));
|
||||
break;
|
||||
case Types.FLOAT:
|
||||
float[] floatColumn = output.getFloatColumn(iCol);
|
||||
System.out.println("Column.length: " + floatColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(floatColumn));
|
||||
break;
|
||||
case Types.SMALLINT:
|
||||
short[] shortColumn = output.getShortColumn(iCol);
|
||||
System.out.println("Column.length: " + shortColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(shortColumn));
|
||||
break;
|
||||
case Types.NVARCHAR:
|
||||
String[] stringColumn = output.getStringColumn(iCol);
|
||||
System.out.println("Column.length: " + stringColumn.length);
|
||||
System.out.println("Column: " + Arrays.toString(stringColumn));
|
||||
break;
|
||||
default:
|
||||
System.out.println("No columnType " + JDBCType.valueOf(columnType).getName());
|
||||
}
|
||||
}
|
||||
|
||||
// cleanup
|
||||
scorer.cleanup();
|
||||
}
|
||||
}
|
||||
+4
@@ -0,0 +1,4 @@
|
||||
age,workclass,fnlwgt,education,education_num,marital_status,occupation,relationship,race,sex,capital_gain,capital_loss,hours_per_week,native_country,income
|
||||
39,State-gov,77516,Bachelors,13,Never-married,Adm-clerical,Not-in-family,White,Male,2174,0,40,United-States,<=50K
|
||||
50,Self-emp-not-inc,83311,Bachelors,13,Married-civ-spouse,Exec-managerial,Husband,White,Male,0,0,13,United-States,<=50K
|
||||
38,Private,215646,HS-grad,9,Divorced,Handlers-cleaners,Not-in-family,White,Male,0,0,40,United-States,<=50K
|
||||
|
+66
@@ -0,0 +1,66 @@
|
||||
package com.microsoft.sqlserver.mleap
|
||||
|
||||
import java.io.File
|
||||
import java.util.logging.Logger
|
||||
import java.util.logging.Level
|
||||
|
||||
import ml.combust.bundle.BundleFile
|
||||
import ml.combust.mleap.runtime.MleapSupport._
|
||||
import ml.combust.mleap.runtime.frame.Transformer
|
||||
import resource._
|
||||
|
||||
class Predictor extends Score {
|
||||
var model: Transformer = null
|
||||
private val LOGGER = Logger.getLogger(classOf[Scorer].getName)
|
||||
|
||||
def init(model_path: String) {
|
||||
LOGGER.info(s"init($model_path)")
|
||||
|
||||
model = (for(bf <- managed(BundleFile(new File(model_path)))) yield {
|
||||
bf.loadMleapBundle()
|
||||
}).tried.flatMap(identity).get.root
|
||||
|
||||
if (LOGGER.getLevel.intValue() <= Level.INFO.intValue()) {
|
||||
println("\nmodel schema fields:")
|
||||
model.schema.fields.zipWithIndex.foreach {
|
||||
case (field, idx) => println(s"$idx $field")
|
||||
}
|
||||
|
||||
println("\nmodel inputSchema fields:")
|
||||
model.inputSchema.fields.zipWithIndex.foreach {
|
||||
case (field, idx) => println(s"$idx $field")
|
||||
}
|
||||
|
||||
println("\nmodel outputSchema fields:")
|
||||
model.outputSchema.fields.zipWithIndex.foreach {
|
||||
case (field, idx) => println(s"$idx $field")
|
||||
}
|
||||
}
|
||||
|
||||
LOGGER.info(s"model loaded...\n")
|
||||
}
|
||||
|
||||
def run(): Unit = {
|
||||
frame_out = model.transform(frame_in).get
|
||||
|
||||
if (LOGGER.getLevel.intValue() <= Level.INFO.intValue()) {
|
||||
println("\noutput schema fields:")
|
||||
frame_out.schema.fields.zipWithIndex.foreach {
|
||||
case (field, idx) => println(s"$idx $field")
|
||||
}
|
||||
}
|
||||
|
||||
//leapFrame2json(frame_out)
|
||||
}
|
||||
|
||||
def select(fieldNames: Array[String]) {
|
||||
if (fieldNames.nonEmpty && fieldNames.length != 1 && fieldNames(0) != "") {
|
||||
val allFieldNames = frame_out.schema.fields.map(_.name)
|
||||
if (!fieldNames.forall(allFieldNames.contains)) {
|
||||
throw new IllegalArgumentException(s"${fieldNames.toList} must be a subset of $allFieldNames")
|
||||
}
|
||||
|
||||
frame_out = frame_out.select(fieldNames: _*).get
|
||||
}
|
||||
}
|
||||
}
|
||||
+235
@@ -0,0 +1,235 @@
|
||||
package com.microsoft.sqlserver.mleap
|
||||
|
||||
import java.io.File
|
||||
import java.sql.{JDBCType, Types}
|
||||
|
||||
import ml.combust.mleap.runtime.MleapSupport._
|
||||
import ml.combust.mleap.runtime.frame.{DefaultLeapFrame, Row}
|
||||
import ml.combust.mleap.runtime.serialization.{BuiltinFormats, FrameReader}
|
||||
import ml.combust.mleap.core.types._
|
||||
import ml.combust.mleap.tensor.{ByteString, DenseTensor, SparseTensor}
|
||||
|
||||
trait Score {
|
||||
|
||||
var frame_in: DefaultLeapFrame = null
|
||||
var frame_out: DefaultLeapFrame = null
|
||||
|
||||
def getScalaType(sqlType: Int): ScalarType = {
|
||||
sqlType match {
|
||||
case Types.BIT => ScalarType.Boolean
|
||||
case Types.TINYINT => ScalarType.Byte
|
||||
case Types.SMALLINT => ScalarType.Short
|
||||
case Types.INTEGER => ScalarType.Int
|
||||
case Types.BIGINT => ScalarType.Long
|
||||
case Types.FLOAT => ScalarType.Float
|
||||
case Types.DOUBLE => ScalarType.Double
|
||||
case Types.NVARCHAR => ScalarType.String
|
||||
case Types.BINARY => ScalarType.ByteString
|
||||
case _ => throw new IllegalArgumentException("unsupported sql type: " + JDBCType.valueOf(sqlType).getName)
|
||||
}
|
||||
}
|
||||
|
||||
def getSqlType(mleapType: BasicType): Int = {
|
||||
mleapType match {
|
||||
case BasicType.Boolean => Types.BIT
|
||||
case BasicType.Byte => Types.TINYINT
|
||||
case BasicType.Short => Types.SMALLINT
|
||||
case BasicType.Int => Types.INTEGER
|
||||
case BasicType.Long => Types.BIGINT
|
||||
case BasicType.Float => Types.FLOAT
|
||||
case BasicType.Double => Types.DOUBLE
|
||||
case BasicType.String => Types.NVARCHAR
|
||||
case BasicType.ByteString => Types.BINARY
|
||||
case _ => throw new IllegalArgumentException("unsupported mleap type: " + mleapType)
|
||||
}
|
||||
}
|
||||
|
||||
def primitiveDataset2leapFrame(input: PrimitiveDataset) {
|
||||
val nCols = input.getColumnCount()
|
||||
val nRows = input.getRowCount(0) // assuming columns have the same length
|
||||
|
||||
// Create a schema.
|
||||
val fields = List.newBuilder[StructField]
|
||||
for (iCol <- 0 until nCols) {
|
||||
fields += StructField(input.getColumnName(iCol), getScalaType(input.getColumnType(iCol)))
|
||||
}
|
||||
val schema = StructType(fields.result).get
|
||||
|
||||
// Create a dataset to contain all of our values
|
||||
val seqBuilder = Seq.newBuilder[Row]
|
||||
for (iRow <- 0 until nRows) {
|
||||
val values = List.newBuilder[Any]
|
||||
for (iCol <- 0 until nCols) {
|
||||
val columnType = input.getColumnType(iCol)
|
||||
values += (columnType match {
|
||||
case Types.BIT => input.getBooleanColumn(iCol)(iRow)
|
||||
case Types.SMALLINT => input.getShortColumn(iCol)(iRow)
|
||||
case Types.INTEGER => input.getIntColumn(iCol)(iRow)
|
||||
case Types.BIGINT => input.getLongColumn(iCol)(iRow)
|
||||
case Types.FLOAT => input.getFloatColumn(iCol)(iRow)
|
||||
case Types.DOUBLE => input.getDoubleColumn(iCol)(iRow)
|
||||
case Types.NVARCHAR => input.getStringColumn(iCol)(iRow)
|
||||
case Types.VARBINARY => input.getBinaryColumn(iCol)(iRow)
|
||||
case Types.DATE => input.getDateColumn(iCol)(iRow)
|
||||
case _ => throw new IllegalArgumentException(s"No BasicType $columnType")
|
||||
})
|
||||
}
|
||||
seqBuilder += Row(values.result: _*)
|
||||
}
|
||||
|
||||
val dataset = seqBuilder.result
|
||||
|
||||
// Create a LeapFrame from the schema and dataset
|
||||
frame_in = DefaultLeapFrame(schema, dataset)
|
||||
}
|
||||
|
||||
def leapFrame2primitiveDataset(output: PrimitiveDataset) {
|
||||
|
||||
val nRows = frame_out.dataset.length
|
||||
var nCols = 0
|
||||
|
||||
val schema = frame_out.schema
|
||||
val fields = schema.fields
|
||||
|
||||
println("\nouput columns:")
|
||||
for (iField <- 0 until fields.length) {
|
||||
val field: StructField = fields(iField)
|
||||
val name = field.name
|
||||
val dataType = field.dataType
|
||||
val base = dataType.base
|
||||
val shape = dataType.shape
|
||||
|
||||
if (shape.isTensor) {
|
||||
val nDims = shape.asInstanceOf[TensorShape].dimensions.get.length
|
||||
frame_out.dataset(0).getTensor(iField) match {
|
||||
case dense: DenseTensor[_] => {
|
||||
println(s"\t$name: DenseTensor[$base]")
|
||||
}
|
||||
case sparse: SparseTensor[_] => {
|
||||
println(s"\t$name: SparseTensor[$base]")
|
||||
}
|
||||
}
|
||||
|
||||
for (iDim <- 0 until nDims) {
|
||||
val nSlots = field.dataType.shape.asInstanceOf[TensorShape].dimensions.get(iDim)
|
||||
|
||||
for (iSlot <- 0 until nSlots) {
|
||||
output.addColumnMetadata(nCols, name + iSlot, getSqlType(base), 0, 0)
|
||||
base match {
|
||||
case BasicType.Boolean => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Boolean](iField).toDense.values(iSlot)).toArray
|
||||
output.addBooleanColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Byte => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Byte](iField).toDense.values(iSlot).toShort).toArray
|
||||
output.addShortColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Short => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Short](iField).toDense.values(iSlot)).toArray
|
||||
output.addShortColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Int => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Int](iField).toDense.values(iSlot)).toArray
|
||||
output.addIntColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Long => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Long](iField).toDense.values(iSlot)).toArray
|
||||
output.addLongColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Float => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Float](iField).toDense.values(iSlot)).toArray
|
||||
output.addFloatColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Double => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[Double](iField).toDense.values(iSlot)).toArray
|
||||
output.addDoubleColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.String => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[String](iField).toDense.values(iSlot)).toArray
|
||||
output.addStringColumn(nCols, outputDataCol)
|
||||
}
|
||||
case BasicType.ByteString => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getTensor[ByteString](iField).toDense.values(iSlot).bytes).toArray
|
||||
output.addBinaryColumn(nCols, outputDataCol)
|
||||
}
|
||||
case _ => throw new IllegalArgumentException(s"No BasicType $base")
|
||||
}
|
||||
|
||||
nCols += 1
|
||||
}
|
||||
}
|
||||
} else {
|
||||
println(s"\t$name: ScalarType.$base")
|
||||
|
||||
output.addColumnMetadata(nCols, name, getSqlType(base), 0, 0)
|
||||
base match {
|
||||
case BasicType.Boolean => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getBool(iField)).toArray
|
||||
output.addBooleanColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Byte => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getByte(iField).toShort).toArray
|
||||
output.addShortColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Short => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getShort(iField)).toArray
|
||||
output.addShortColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Int => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getInt(iField)).toArray
|
||||
output.addIntColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Long => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getLong(iField)).toArray
|
||||
output.addLongColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Float => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getFloat(iField)).toArray
|
||||
output.addFloatColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.Double => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getDouble(iField)).toArray
|
||||
output.addDoubleColumn(nCols, outputDataCol, null)
|
||||
}
|
||||
case BasicType.String => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getString(iField)).toArray
|
||||
output.addStringColumn(nCols, outputDataCol)
|
||||
}
|
||||
case BasicType.ByteString => {
|
||||
val outputDataCol = frame_out.dataset.map(_.getByteString(iField).bytes).toArray
|
||||
output.addBinaryColumn(nCols, outputDataCol)
|
||||
}
|
||||
case _ => throw new IllegalArgumentException(s"No BasicType $base")
|
||||
}
|
||||
nCols += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
def json2leapFrame(frame_path: String) {
|
||||
println (s"run($frame_path)")
|
||||
|
||||
val f = new File (frame_path)
|
||||
if (f.exists () && ! f.isDirectory () ) {
|
||||
// get input from file
|
||||
frame_in = FrameReader (BuiltinFormats.json).read (f).get
|
||||
} else {
|
||||
// get input from string
|
||||
frame_in = FrameReader (BuiltinFormats.json).fromBytes (frame_path.getBytes () ).get
|
||||
}
|
||||
}
|
||||
|
||||
def leapFrame2json(frame: DefaultLeapFrame): String = {
|
||||
var json_str: String = null
|
||||
for(bytes <- frame.writer("ml.combust.mleap.json").toBytes();
|
||||
frame2 <- FrameReader("ml.combust.mleap.json").fromBytes(bytes)) {
|
||||
json_str = new String(bytes)
|
||||
assert(frame == frame2)
|
||||
}
|
||||
|
||||
println()
|
||||
println(json_str)
|
||||
|
||||
return json_str
|
||||
}
|
||||
}
|
||||
+98
@@ -0,0 +1,98 @@
|
||||
package com.microsoft.sqlserver.mleap;
|
||||
|
||||
import org.junit.*;
|
||||
|
||||
import java.sql.Types;
|
||||
import java.util.*;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class ScorerTest {
|
||||
|
||||
private static PrimitiveDataset input = new PrimitiveDataset();
|
||||
private static LinkedHashMap<String, Object> params = new LinkedHashMap<>();
|
||||
private static PrimitiveDataset output;
|
||||
|
||||
@BeforeClass
|
||||
public static void score() {
|
||||
// get model and testing data
|
||||
String modelPath = "src/main/resources/adult_census_pipeline.zip";
|
||||
|
||||
int columnId = 0;
|
||||
input.addColumnMetadata(columnId, "age", java.sql.Types.INTEGER, 0, 0);
|
||||
input.addIntColumn(columnId, new int[]{39, 50, 38}, null);
|
||||
|
||||
columnId++;
|
||||
input.addColumnMetadata(columnId, "hours_per_week", java.sql.Types.INTEGER, 0, 0);
|
||||
input.addIntColumn(columnId, new int[]{40, 13, 40}, null);
|
||||
|
||||
columnId++;
|
||||
input.addColumnMetadata(columnId, "education", Types.NVARCHAR, 0, 0);
|
||||
input.addStringColumn(columnId, new String[]{"Bachelors", "Bachelors", "HS-grad"});
|
||||
|
||||
columnId++;
|
||||
input.addColumnMetadata(columnId, "sex", Types.NVARCHAR, 0, 0);
|
||||
input.addStringColumn(columnId, new String[]{"Male", "Male", "Male"});
|
||||
|
||||
columnId++;
|
||||
input.addColumnMetadata(columnId, "income", Types.NVARCHAR, 0, 0);
|
||||
input.addStringColumn(columnId, new String[]{"<=50K", "<=50K", "<=50K"});
|
||||
|
||||
// specify some params
|
||||
params.put("logLevel", "INFO"); //default WARN
|
||||
params.put("modelPath", modelPath);
|
||||
params.put("outputFields", "prediction,probability,education,sex,income,predictedIncome");
|
||||
//params.put("outputFields", "features,education-encoded"); //SparseTensor
|
||||
|
||||
// perform scoring
|
||||
Scorer scorer = new Scorer();
|
||||
scorer.init("session0", 0, 1);
|
||||
output = scorer.execute(input, params);
|
||||
|
||||
// cleanup
|
||||
scorer.cleanup();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputColumnCountShouldMatch() {
|
||||
// display output
|
||||
int nOutputCols = output.getColumnCount();
|
||||
int nOutputFields = params.getOrDefault("outputFields", "").toString().split(",").length;
|
||||
assertEquals(nOutputFields + 1, nOutputCols); // "probability" is a vector field of size 2 in this case
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputColumnNamesShouldMatch() {
|
||||
int nOutputCols = output.getColumnCount();
|
||||
|
||||
Set<String> columnNames = new HashSet<>();
|
||||
for (int iCol = 0; iCol < nOutputCols; iCol++) {
|
||||
columnNames.add(output.getColumnName(iCol));
|
||||
}
|
||||
Set<String> fieldNames = new HashSet<>(Arrays.asList("prediction","probability0","probability1","education","sex","income","predictedIncome"));
|
||||
assertEquals(fieldNames, columnNames);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputColumnValuesShouldMatch() {
|
||||
String columnName = "education";
|
||||
|
||||
int outputIndex = output.getColumnIndex(columnName);
|
||||
String[] outputStringColumn = output.getStringColumn(outputIndex);
|
||||
|
||||
int inputIndex = input.getColumnIndex(columnName);
|
||||
String[] inputStringColumn = input.getStringColumn(inputIndex);
|
||||
|
||||
assertArrayEquals(inputStringColumn, outputStringColumn);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void outputRowCountShouldMatch() {
|
||||
int rowCount0 = input.getRowCount(0);
|
||||
int[] rowCounts = output.getRowCounts();
|
||||
|
||||
for (int rowCount: rowCounts) {
|
||||
assertEquals(rowCount, rowCount0);
|
||||
}
|
||||
}
|
||||
}
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
package com.microsoft.sqlserver.mleap
|
||||
|
||||
import java.sql.Types
|
||||
|
||||
import org.scalatest.fixture
|
||||
|
||||
class PredictorTest extends fixture.FlatSpec {
|
||||
|
||||
case class FixtureParam(input: PrimitiveDataset, scorer: Predictor, output: PrimitiveDataset)
|
||||
|
||||
def withFixture(test: OneArgTest) = {
|
||||
val input: PrimitiveDataset = new PrimitiveDataset
|
||||
val scorer = new Predictor
|
||||
val output: PrimitiveDataset = new PrimitiveDataset
|
||||
val theFixture = FixtureParam(input, scorer, output)
|
||||
|
||||
try {
|
||||
var columnId = 0
|
||||
input.addColumnMetadata(columnId, "age", java.sql.Types.INTEGER, 0, 0)
|
||||
input.addIntColumn(columnId, Array[Int](39, 50, 38), null)
|
||||
|
||||
columnId += 1
|
||||
input.addColumnMetadata(columnId, "hours_per_week", java.sql.Types.INTEGER, 0, 0)
|
||||
input.addIntColumn(columnId, Array[Int](40, 13, 40), null)
|
||||
|
||||
columnId += 1
|
||||
input.addColumnMetadata(columnId, "education", Types.NVARCHAR, 0, 0)
|
||||
input.addStringColumn(columnId, Array[String]("Bachelors", "Bachelors", "HS-grad"))
|
||||
|
||||
columnId += 1
|
||||
input.addColumnMetadata(columnId, "sex", Types.NVARCHAR, 0, 0)
|
||||
input.addStringColumn(columnId, Array[String]("Male", "Male", "Male"))
|
||||
|
||||
columnId += 1
|
||||
input.addColumnMetadata(columnId, "income", Types.NVARCHAR, 0, 0)
|
||||
input.addStringColumn(columnId, Array[String]("<=50K", "<=50K", "<=50K"))
|
||||
|
||||
scorer.primitiveDataset2leapFrame(input)
|
||||
scorer.frame_out = scorer.frame_in
|
||||
scorer.leapFrame2primitiveDataset(output)
|
||||
|
||||
withFixture(test.toNoArgTest(theFixture)) // "loan" the fixture to the test
|
||||
}
|
||||
finally () // clean up the fixture, nothing in this case
|
||||
}
|
||||
|
||||
"A Predictor" should "be able to convert PrimitiveDataset to DefaultLeapFrame with same field names" in { f =>
|
||||
|
||||
val columnNames = f.input.getColumnNames()
|
||||
val fieldNames = f.scorer.frame_in.schema.fields.map(_.name).toArray
|
||||
assert(fieldNames.deep == columnNames.deep)
|
||||
}
|
||||
|
||||
it should "be able to convert DefaultLeapFrame to PrimitiveDataset with same column names" in { f =>
|
||||
|
||||
val columnNames = f.output.getColumnNames()
|
||||
val fieldNames = f.scorer.frame_out.schema.fields.map(_.name).toArray
|
||||
assert(fieldNames.deep == columnNames.deep)
|
||||
}
|
||||
|
||||
it should "be able to convert PrimitiveDataset to DefaultLeapFrame with same int field values" in { f =>
|
||||
|
||||
val columnName = "age"
|
||||
val columnIndex = f.input.getColumnIndex(columnName)
|
||||
val columnValues = f.input.getIntColumn(columnIndex)
|
||||
|
||||
val fieldNames = f.scorer.frame_in.schema.fields.map(_.name).toArray
|
||||
val iField = fieldNames.indexOf(columnName)
|
||||
val fieldValues = f.scorer.frame_in.dataset.map(_.getInt(iField)).toArray
|
||||
|
||||
assert(fieldValues.deep == columnValues.deep)
|
||||
}
|
||||
|
||||
it should "be able to convert DefaultLeapFrame to PrimitiveDataset with same int column values" in { f =>
|
||||
|
||||
val columnName = "age"
|
||||
val columnIndex = f.input.getColumnIndex(columnName)
|
||||
val columnValues = f.input.getIntColumn(columnIndex)
|
||||
|
||||
val fieldNames = f.scorer.frame_out.schema.fields.map(_.name).toArray
|
||||
val iField = fieldNames.indexOf(columnName)
|
||||
val fieldValues = f.scorer.frame_out.dataset.map(_.getInt(iField)).toArray
|
||||
|
||||
assert(fieldValues.deep == columnValues.deep)
|
||||
}
|
||||
|
||||
it should "be able to convert PrimitiveDataset to DefaultLeapFrame with same string field values" in { f =>
|
||||
|
||||
val columnName = "education"
|
||||
val columnIndex = f.input.getColumnIndex(columnName)
|
||||
val columnValues = f.input.getStringColumn(columnIndex)
|
||||
|
||||
val fieldNames = f.scorer.frame_in.schema.fields.map(_.name).toArray
|
||||
val iField = fieldNames.indexOf(columnName)
|
||||
val fieldValues = f.scorer.frame_in.dataset.map(_.getString(iField)).toArray
|
||||
|
||||
assert(fieldValues.deep == columnValues.deep)
|
||||
}
|
||||
|
||||
it should "be able to convert DefaultLeapFrame to PrimitiveDataset with same string column values" in { f =>
|
||||
|
||||
val columnName = "education"
|
||||
val columnIndex = f.input.getColumnIndex(columnName)
|
||||
val columnValues = f.input.getStringColumn(columnIndex)
|
||||
|
||||
val fieldNames = f.scorer.frame_out.schema.fields.map(_.name).toArray
|
||||
val iField = fieldNames.indexOf(columnName)
|
||||
val fieldValues = f.scorer.frame_out.dataset.map(_.getString(iField)).toArray
|
||||
|
||||
assert(fieldValues.deep == columnValues.deep)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user