diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/README.md b/samples/features/sql-big-data-cluster/spark/sparkml/README.md
new file mode 100644
index 00000000..7965e6fc
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/README.md
@@ -0,0 +1,18 @@
+# MLeap on SQL Server Big Data cluster
+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](mleap-docs.combust.ml/), 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)
+
+![Train_Score_Export_with_Spark.jpg](Train_Score_Export_with_Spark.jpg)
+
+## 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).
+
+An equivalent Jupyter notebook is also included [here](train_score_export_ml_models_with_spark.ipynb) if it is preferred over pure Python code.
+
+## 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).
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg b/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg
index 0cc016f8..5893e55c 100644
Binary files a/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg and b/samples/features/sql-big-data-cluster/spark/sparkml/Train_Score_Export_with_Spark.jpg differ
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/cleanup.sh b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/cleanup.sh
new file mode 100644
index 00000000..fb8dcd10
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/cleanup.sh
@@ -0,0 +1,6 @@
+#!/bin/bash
+
+echo "Cleaning up mleap_sql tests"
+
+hadoop fs -rm /user/root/AdultCensusIncome.csv
+rm AdultCensusIncome.csv
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_pyspark.py b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_pyspark.py
new file mode 100644
index 00000000..c450d166
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_pyspark.py
@@ -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))
+
+###############################################################################
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_sql_tests.py b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_sql_tests.py
new file mode 100644
index 00000000..a1b3e1c8
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/mleap_sql_tests.py
@@ -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'}]
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/setup.sh b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/setup.sh
new file mode 100644
index 00000000..80f376c3
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/setup.sh
@@ -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/
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/test.sh b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/test.sh
new file mode 100644
index 00000000..7308e25f
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mleap_sql_test/test.sh
@@ -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
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/Makefile b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/Makefile
new file mode 100644
index 00000000..5f7abc4e
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/Makefile
@@ -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
+
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/build.sbt b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/build.sbt
new file mode 100644
index 00000000..ef27c126
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/build.sbt
@@ -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"}
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/build.properties b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/build.properties
new file mode 100644
index 00000000..e9a676f7
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/build.properties
@@ -0,0 +1 @@
+sbt.version = 1.1.5
\ No newline at end of file
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/plugins.sbt b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/plugins.sbt
new file mode 100644
index 00000000..652a3b93
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/project/plugins.sbt
@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.6")
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/PrimitiveDataset.java b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/PrimitiveDataset.java
new file mode 100644
index 00000000..3e9a269d
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/PrimitiveDataset.java
@@ -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;
+    }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/Scorer.java b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/Scorer.java
new file mode 100644
index 00000000..a7f60117
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/java/com/microsoft/sqlserver/mleap/Scorer.java
@@ -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();
+    }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/resources/adult_census_income.csv b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/resources/adult_census_income.csv
new file mode 100644
index 00000000..48a1b390
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/resources/adult_census_income.csv
@@ -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
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Predictor.scala b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Predictor.scala
new file mode 100644
index 00000000..c319a510
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Predictor.scala
@@ -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
+    }
+  }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Score.scala b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Score.scala
new file mode 100644
index 00000000..342dcfa7
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/main/scala/com/microsoft/sqlserver/mleap/Score.scala
@@ -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
+  }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/java/com/microsoft/sqlserver/mleap/ScorerTest.java b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/java/com/microsoft/sqlserver/mleap/ScorerTest.java
new file mode 100644
index 00000000..9f4e4455
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/java/com/microsoft/sqlserver/mleap/ScorerTest.java
@@ -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);
+        }
+    }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/scala/com/microsoft/sqlserver/mleap/PredictorTest.scala b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/scala/com/microsoft/sqlserver/mleap/PredictorTest.scala
new file mode 100644
index 00000000..b98ddcbe
--- /dev/null
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/mssql-mleap-app/src/test/scala/com/microsoft/sqlserver/mleap/PredictorTest.scala
@@ -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)
+  }
+}
diff --git a/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb b/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb
index df4bd268..c3219ad8 100644
--- a/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb
+++ b/samples/features/sql-big-data-cluster/spark/sparkml/train_score_export_ml_models_with_spark.ipynb
@@ -19,12 +19,12 @@
     "cells": [
         {
             "cell_type": "markdown",
-            "source": "# Machine learning with SPARK in SQL Server 2019 Big Data Cluster\r\nSpark in Unified Big data compute engine that enables big data processing, Machine learning and AI\r\n\r\nKey Spark advantages are \r\n1. Distributed compute enging \r\n2. Choice of langauge (Python, R, Scala, Java)\r\n3. Single engine for Batch and Streaming jobs\r\n\r\nIn this tutorial we'll cover how we can use Spark to create and deploy machine learning models. The example is a python(PySpark) sample. The same can also be done using Scala and R ( SparkR) in Spark.\r\n\r\n<img src = \"C:\\repos\\sql-server-samples\\samples\\features\\sql-big-data-cluster\\spark\\sparkml\\Train_Score_Export_with_Spark.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\r\n\r\n## Steps\r\n1. Explore your Data\r\n2. Data Prep and split Data as Training and Test set\r\n3. Model Training\r\n4. Model Scoring \r\n5. Persist as Spark Model\r\n6. Persist as Portable Model\r\n\r\nE2E machine learning involves several additional step e.g data exploration, feature selection and principal component analysis,model selection etc. Many of these steps are ignored here for brevity.\r\n\r\n\r\n\r\n",
+            "source": "# Machine learning with SPARK in SQL Server 2019 Big Data Cluster\r\nSpark in Unified Big data compute engine that enables big data processing, Machine learning and AI\r\n\r\nKey Spark advantages are \r\n1. Distributed compute enging \r\n2. Choice of langauge (Python, R, Scala, Java)\r\n3. Single engine for Batch and Streaming jobs\r\n\r\nIn this tutorial we'll cover how we can use Spark to create and deploy machine learning models. The example is a python(PySpark) sample. The same can also be done using Scala and R ( SparkR) in Spark.\r\n\r\n<img src = \"Train_Score_Export_with_Spark.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\r\n\r\n## Steps\r\n1. Explore your Data\r\n2. Data Prep and split Data as Training and Test set\r\n3. Model Training\r\n4. Model Scoring \r\n5. Persist as Spark Model\r\n6. Persist as Portable Model\r\n\r\nE2E machine learning involves several additional step e.g data exploration, feature selection and principal component analysis,model selection etc. Many of these steps are ignored here for brevity.\r\n\r\n\r\n\r\n",
             "metadata": {}
         },
         {
             "cell_type": "markdown",
-            "source": "## Step 1 - Explore your data\r\n### Load the data\r\nFor this example we'll use **AdultCensusIncome** data from [here]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ). From your Azure Data Studio connect to the HDFS/Spark gateway and create a directory called spark_data under HDFS. \r\nDownload [AdultCensusIncome.csv]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ) to your local machine and upload to HDFS.Upload AdultCensusIncome.csv to the folder we created.\r\n\r\n### Exploratory Analysis\r\n- Baisc exploration on the data\r\n- Labels & Features\r\n1. **Label**    - This refers to predicted value. This is represented as a column in the data. Label is **income** \r\n2. **Features** - This refers to the characteristics that are used to predict. **age** and **hours_per_week**\r\n\r\nNote : In reality features are chosen by applying some correlations techniques to understand what best characterize the Label we are predicting.\r\n\r\n### The Model we will build\r\nIn AdultCensusIncome.csv contains several columsn like Income range, age, hours-per-week, education, occupation etc. We'll build a model that can predict income range would be >50K or <50K.\r\n",
+            "source": "## Step 1 - Explore your data\r\n### Load the data\r\nFor this example we'll use **AdultCensusIncome** data from [here]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ). From your Azure Data Studio connect to the HDFS/Spark gateway and create a directory called spark_data under HDFS. \r\nDownload [AdultCensusIncome.csv]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ) to your local machine and upload to HDFS.Upload AdultCensusIncome.csv to the folder we created.\r\n\r\n### Exploratory Analysis\r\n- Baisc exploration on the data\r\n- Labels & Features\r\n1. **Label**    - This refers to predicted value. This is represented as a column in the data. Label is **income** \r\n2. **Features** - This refers to the characteristics that are used to predict. **age**, **hours_per_week**, and **education**\r\n\r\nNote : In reality features are chosen by applying some correlations techniques to understand what best characterize the Label we are predicting.\r\n\r\n### The Model we will build\r\nIn AdultCensusIncome.csv contains several columsn like Income range, age, hours-per-week, education, occupation etc. We'll build a model that can predict income range would be >50K or <50K.\r\n",
             "metadata": {}
         },
         {
@@ -33,39 +33,57 @@
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "Number of rows: 32561,  Number of coulumns : 15\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education-num: integer (nullable = true)\n |-- marital-status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital-gain: integer (nullable = true)\n |-- capital-loss: integer (nullable = true)\n |-- hours-per-week: integer (nullable = true)\n |-- native-country: string (nullable = true)\n |-- income: string (nullable = true)\n\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education_num: integer (nullable = true)\n |-- marital_status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital_gain: integer (nullable = true)\n |-- capital_loss: integer (nullable = true)\n |-- hours_per_week: integer (nullable = true)\n |-- native_country: string (nullable = true)\n |-- income: string (nullable = true)"
+                    "text": "Starting Spark application\n",
+                    "output_type": "stream"
+                },
+                {
+                    "data": {
+                        "text/plain": "<IPython.core.display.HTML object>",
+                        "text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>20</td><td>application_1559313998190_0086</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"https://10.193.17.7:30443/gateway/default/yarn/proxy/application_1559313998190_0086/\">Link</a></td><td><a target=\"_blank\" href=\"https://10.193.17.7:30443/gateway/default/yarn/container/container_1559313998190_0086_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
+                    },
+                    "metadata": {},
+                    "output_type": "display_data"
+                },
+                {
+                    "name": "stdout",
+                    "text": "SparkSession available as 'spark'.\n",
+                    "output_type": "stream"
+                },
+                {
+                    "name": "stdout",
+                    "text": "Number of rows: 32561,  Number of coulumns : 15\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education-num: integer (nullable = true)\n |-- marital-status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital-gain: integer (nullable = true)\n |-- capital-loss: integer (nullable = true)\n |-- hours-per-week: integer (nullable = true)\n |-- native-country: string (nullable = true)\n |-- income: string (nullable = true)\n\nroot\n |-- age: integer (nullable = true)\n |-- workclass: string (nullable = true)\n |-- fnlwgt: integer (nullable = true)\n |-- education: string (nullable = true)\n |-- education_num: integer (nullable = true)\n |-- marital_status: string (nullable = true)\n |-- occupation: string (nullable = true)\n |-- relationship: string (nullable = true)\n |-- race: string (nullable = true)\n |-- sex: string (nullable = true)\n |-- capital_gain: integer (nullable = true)\n |-- capital_loss: integer (nullable = true)\n |-- hours_per_week: integer (nullable = true)\n |-- native_country: string (nullable = true)\n |-- income: string (nullable = true)",
+                    "output_type": "stream"
+                }
+            ],
+            "execution_count": 2
+        },
+        {
+            "cell_type": "code",
+            "source": "#Basic data exploration\r\n\r\n##1. Sub set the data and print some important columns\r\nprint(\"Select few columns to see the data\")\r\ndata_all.select(['income','age','hours_per_week', 'education']).show(10)\r\n\r\n## Find the number of distict values\r\nprint(\"Number of distinct values for income\")\r\nds_sub = data_all.select('income').distinct()\r\nds_sub.show()\r\n\r\n##Add a numberic column(income_code) derived from income column\r\nprint(\"Added numeric column(income_code) derived from income column\")\r\nfrom pyspark.sql.functions import expr\r\n\r\ndf_new = data_all.withColumn(\"income_code\", expr(\"case \\\r\n                                            when income like '%<=50K%' then 0 \\\r\n                                            when income like '%>50K%' then 1 \\\r\n                                            else 2 end \"))\r\n\r\ndf_new.select(['income', 'age', 'hours_per_week', 'education', 'income_code']).show(10)\r\n\r\n##Summary  statistical operations on dataframe\r\nprint(\"Print a statistical summary of a few columns\")\r\ndf_new.select(['income','age','hours_per_week', 'education','income_code']).describe().show()\r\n\r\nprint(\"Calculate Co variance between a few columns to understand features to use\")\r\nmycov = df_new.stat.cov('income_code','hours_per_week')\r\nprint(\"Covariance between income and hours_per_week is\", round(mycov,1))\r\n\r\nmycov = df_new.stat.cov('income_code','age')\r\nprint(\"Covariance between income and age is\", round(mycov,1))\r\n\r\n",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stdout",
+                    "text": "Select few columns to see the data\n+------+---+--------------+---------+\n|income|age|hours_per_week|education|\n+------+---+--------------+---------+\n| <=50K| 39|            40|Bachelors|\n| <=50K| 50|            13|Bachelors|\n| <=50K| 38|            40|  HS-grad|\n| <=50K| 53|            40|     11th|\n| <=50K| 28|            40|Bachelors|\n| <=50K| 37|            40|  Masters|\n| <=50K| 49|            16|      9th|\n|  >50K| 52|            45|  HS-grad|\n|  >50K| 31|            50|  Masters|\n|  >50K| 42|            40|Bachelors|\n+------+---+--------------+---------+\nonly showing top 10 rows\n\nNumber of distinct values for income\n+------+\n|income|\n+------+\n| <=50K|\n|  >50K|\n+------+\n\nAdded numeric column(income_code) derived from income column\n+------+---+--------------+---------+-----------+\n|income|age|hours_per_week|education|income_code|\n+------+---+--------------+---------+-----------+\n| <=50K| 39|            40|Bachelors|          0|\n| <=50K| 50|            13|Bachelors|          0|\n| <=50K| 38|            40|  HS-grad|          0|\n| <=50K| 53|            40|     11th|          0|\n| <=50K| 28|            40|Bachelors|          0|\n| <=50K| 37|            40|  Masters|          0|\n| <=50K| 49|            16|      9th|          0|\n|  >50K| 52|            45|  HS-grad|          1|\n|  >50K| 31|            50|  Masters|          1|\n|  >50K| 42|            40|Bachelors|          1|\n+------+---+--------------+---------+-----------+\nonly showing top 10 rows\n\nPrint a statistical summary of a few columns\n+-------+------+------------------+------------------+------------+-------------------+\n|summary|income|               age|    hours_per_week|   education|        income_code|\n+-------+------+------------------+------------------+------------+-------------------+\n|  count| 32561|             32561|             32561|       32561|              32561|\n|   mean|  null| 38.58164675532078|40.437455852092995|        null| 0.2408095574460244|\n| stddev|  null|13.640432553581356|12.347428681731838|        null|0.42758148856469247|\n|    min| <=50K|                17|                 1|        10th|                  0|\n|    max|  >50K|                90|                99|Some-college|                  1|\n+-------+------+------------------+------------------+------------+-------------------+\n\nCalculate Co variance between a few columns to understand features to use\nCovariance between income and hours_per_week is 1.2\nCovariance between income and age is 1.4",
+                    "output_type": "stream"
                 }
             ],
             "execution_count": 3
         },
         {
             "cell_type": "code",
-            "source": "#Basic data exploration\r\n\r\n##1. Sub set the data and print some important columns\r\nprint(\"Select few columns to see the data\")\r\ndata_all.select(['income','age','hours_per_week']).show(10)\r\n\r\n## Find the number of distict values\r\nprint(\"Number of distinct values for income\")\r\nds_sub = data_all.select('income').distinct()\r\nds_sub.show()\r\n\r\n##Add a numberic column(income_code) derived from income column\r\nprint(\"Added numeric column(income_code) derived from income column\")\r\nfrom pyspark.sql.functions import expr\r\n\r\ndf_new = data_all.withColumn(\"income_code\", expr(\"case \\\r\n                                            when income like '%<=50K%' then 0 \\\r\n                                            when income like '%>50K%' then 1 \\\r\n                                            else 2 end \"))\r\n\r\ndf_new.select(['income','age','hours_per_week','income_code']).show(10)\r\n\r\n##Summary  statistical operations on dataframe\r\nprint(\"Print a statistical summary of a few columns\")\r\ndf_new.select(['income','age','hours_per_week','income_code']).describe().show()\r\n\r\nprint(\"Calculate Co variance between a few columns to understand features to use\")\r\nmycov = df_new.stat.cov('income_code','hours_per_week')\r\nprint(\"Covariance between income and hours_per_week is\", round(mycov,1))\r\n\r\nmycov = df_new.stat.cov('income_code','age')\r\nprint(\"Covariance between income and age is\", round(mycov,1))\r\n\r\n",
+            "source": "# Choose feature columns and the label column.\r\nlabel = \"income\"\r\nxvars = [\"age\", \"hours_per_week\", 'education'] #numeric and string\r\n\r\nprint(\"label = {}\".format(label))\r\nprint(\"features = {}\".format(xvars))\r\n\r\n#Check label counts to check data bias\r\nprint(\"Count of rows that are <=50K\", data_all[data_all.income==\"<=50K\"].count())\r\nprint(\"Count of rows that are >50K\", data_all[data_all.income==\">50K\"].count())\r\n\r\n\r\nselect_cols = xvars\r\nselect_cols.append(label)\r\ndata = data_all.select(select_cols)",
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "Select few columns to see the data\n+------+---+--------------+\n|income|age|hours_per_week|\n+------+---+--------------+\n| <=50K| 39|            40|\n| <=50K| 50|            13|\n| <=50K| 38|            40|\n| <=50K| 53|            40|\n| <=50K| 28|            40|\n| <=50K| 37|            40|\n| <=50K| 49|            16|\n|  >50K| 52|            45|\n|  >50K| 31|            50|\n|  >50K| 42|            40|\n+------+---+--------------+\nonly showing top 10 rows\n\nNumber of distinct values for income\n+------+\n|income|\n+------+\n| <=50K|\n|  >50K|\n+------+\n\nAdded numeric column(income_code) derived from income column\n+------+---+--------------+-----------+\n|income|age|hours_per_week|income_code|\n+------+---+--------------+-----------+\n| <=50K| 39|            40|          0|\n| <=50K| 50|            13|          0|\n| <=50K| 38|            40|          0|\n| <=50K| 53|            40|          0|\n| <=50K| 28|            40|          0|\n| <=50K| 37|            40|          0|\n| <=50K| 49|            16|          0|\n|  >50K| 52|            45|          1|\n|  >50K| 31|            50|          1|\n|  >50K| 42|            40|          1|\n+------+---+--------------+-----------+\nonly showing top 10 rows\n\nPrint a statistical summary of a few columns\n+-------+------+------------------+------------------+-------------------+\n|summary|income|               age|    hours_per_week|        income_code|\n+-------+------+------------------+------------------+-------------------+\n|  count| 32561|             32561|             32561|              32561|\n|   mean|  null| 38.58164675532078|40.437455852092995| 0.2408095574460244|\n| stddev|  null|13.640432553581356|12.347428681731838|0.42758148856469247|\n|    min| <=50K|                17|                 1|                  0|\n|    max|  >50K|                90|                99|                  1|\n+-------+------+------------------+------------------+-------------------+\n\nCalculate Co variance between a few columns to understand features to use\nCovariance between income and hours_per_week is 1.2\nCovariance between income and age is 1.4"
+                    "text": "label = income\nfeatures = ['age', 'hours_per_week', 'education']\nCount of rows that are <=50K 24720\nCount of rows that are >50K 7841",
+                    "output_type": "stream"
                 }
             ],
             "execution_count": 4
         },
-        {
-            "cell_type": "code",
-            "source": "# Choose feature columns and the label column.\r\nlabel = \"income\"\r\nxvars = [\"age\", \"hours_per_week\"] #all numeric\r\n\r\nprint(\"label = {}\".format(label))\r\nprint(\"features = {}\".format(xvars))\r\n\r\n#Check label counts to check data bias\r\nprint(\"Count of rows that are <=50K\", data_all[data_all.income==\"<=50K\"].count())\r\nprint(\"Count of rows that are >50K\", data_all[data_all.income==\">50K\"].count())\r\n\r\n\r\nselect_cols = xvars\r\nselect_cols.append(label)\r\ndata = data_all.select(select_cols)",
-            "metadata": {},
-            "outputs": [
-                {
-                    "output_type": "stream",
-                    "name": "stdout",
-                    "text": "label = income\nfeatures = ['age', 'hours_per_week']\nCount of rows that are <=50K 24720\nCount of rows that are >50K 7841"
-                }
-            ],
-            "execution_count": 5
-        },
         {
             "cell_type": "markdown",
             "source": "## Step 2 - Split as training and test set\r\nWe'll use 75% of rows to train the model and rest of the 25% to evaluate the model. Additionally we persist the train and test data sets to HDFS storage. The step is not necessary , but shown to demonstrate saving and loading with ORC format. Other format e.g. Parquet may also be used. Post this step you should see 2 directories created with the name \"AdultCensusIncomeTrain\" and \"AdultCensusIncomeTest\"\r\n",
@@ -77,30 +95,30 @@
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "train (24469, 3)\ntest (8092, 3)\ntrain and test datasets saved to /spark_ml/AdultCensusIncomeTrain and /spark_ml/AdultCensusIncomeTest"
+                    "text": "train (24469, 4)\ntest (8092, 4)\ntrain and test datasets saved to /spark_ml/AdultCensusIncomeTrain and /spark_ml/AdultCensusIncomeTest",
+                    "output_type": "stream"
                 }
             ],
-            "execution_count": 6
+            "execution_count": 5
         },
         {
             "cell_type": "markdown",
-            "source": "## Step 3 - Train a model\r\n[Spark ML pipeline] ( https://spark.apache.org/docs/2.3.1/ml-pipeline.html ) allow to sequence all steps as a workflow and make it easier to experiment with various algorithms and their parameters. The following code first constructs the stages and then puts these stages together in Ml pipeline.  LogisticRegression is used to create the model.\r\n\r\n",
+            "source": "## Step 3 - Train a model\r\n[Spark ML pipelines](https://spark.apache.org/docs/latest/ml-pipeline.html) allow to sequence all steps as a workflow and make it easier to experiment with various algorithms and their parameters. The following code first constructs the stages and then puts these stages together in Ml pipeline.  LogisticRegression is used to create the model.",
             "metadata": {}
         },
         {
             "cell_type": "code",
-            "source": "from pyspark.ml import Pipeline, PipelineModel\r\nfrom pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler\r\nfrom pyspark.ml.classification import LogisticRegression\r\n\r\nreg = 0.1\r\nprint(\"Using LogisticRegression model with Regularization Rate of {}.\".format(reg))\r\n\r\n# create a new Logistic Regression model.\r\nlr = LogisticRegression(regParam=reg)\r\n\r\ndtypes = dict(train.dtypes)\r\ndtypes.pop(label)\r\n\r\nsi_xvars = []\r\nohe_xvars = []\r\nfeatureCols = []\r\nfor idx,key in enumerate(dtypes):\r\n    if dtypes[key] == \"string\":\r\n        featureCol = \"-\".join([key, \"encoded\"])\r\n        featureCols.append(featureCol)\r\n        \r\n        tmpCol = \"-\".join([key, \"tmp\"])\r\n        si_xvars.append(StringIndexer(inputCol=key, outputCol=tmpCol, handleInvalid=\"skip\")) #, handleInvalid=\"keep\"\r\n        ohe_xvars.append(OneHotEncoder(inputCol=tmpCol, outputCol=featureCol))\r\n    else:\r\n        featureCols.append(key)\r\n\r\n# string-index the label column into a column named \"label\"\r\nsi_label = StringIndexer(inputCol=label, outputCol='label')\r\n\r\n# assemble the encoded feature columns in to a column named \"features\"\r\nassembler = VectorAssembler(inputCols=featureCols, outputCol=\"features\")\r\n\r\n\r\nstages = []\r\nstages.extend(si_xvars)\r\nstages.extend(ohe_xvars)\r\nstages.append(si_label)\r\nstages.append(assembler)\r\nstages.append(lr)\r\npipe = Pipeline(stages=stages)\r\nprint(\"Pipeline Created\")\r\n\r\nmodel = pipe.fit(train)\r\nprint(\"Model Trained\")\r\nprint(\"Model is \", model)\r\nprint(\"Model Stages\", model.stages)",
+            "source": "from pyspark.ml import Pipeline, PipelineModel\r\nfrom pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler\r\nfrom pyspark.ml.classification import LogisticRegression\r\n\r\nreg = 0.1\r\nprint(\"Using LogisticRegression model with Regularization Rate of {}.\".format(reg))\r\n\r\n# create a new Logistic Regression model.\r\nlr = LogisticRegression(regParam=reg)\r\n\r\ndtypes = dict(train.dtypes)\r\ndtypes.pop(label)\r\n\r\nsi_xvars = []\r\nohe_xvars = []\r\nfeatureCols = []\r\nfor idx,key in enumerate(dtypes):\r\n    if dtypes[key] == \"string\":\r\n        featureCol = \"-\".join([key, \"encoded\"])\r\n        featureCols.append(featureCol)\r\n        \r\n        tmpCol = \"-\".join([key, \"tmp\"])\r\n        si_xvars.append(StringIndexer(inputCol=key, outputCol=tmpCol, handleInvalid=\"skip\")) #, handleInvalid=\"keep\"\r\n        ohe_xvars.append(OneHotEncoderEstimator(inputCols=[tmpCol], outputCols=[featureCol]))\r\n    else:\r\n        featureCols.append(key)\r\n\r\n# string-index the label column into a column named \"label\"\r\nsi_label = StringIndexer(inputCol=label, outputCol='label')\r\n\r\n# assemble the encoded feature columns in to a column named \"features\"\r\nassembler = VectorAssembler(inputCols=featureCols, outputCol=\"features\")\r\n\r\n\r\nstages = []\r\nstages.extend(si_xvars)\r\nstages.extend(ohe_xvars)\r\nstages.append(si_label)\r\nstages.append(assembler)\r\nstages.append(lr)\r\npipe = Pipeline(stages=stages)\r\nprint(\"Pipeline Created\")\r\n\r\nmodel = pipe.fit(train)\r\nprint(\"Model Trained\")\r\nprint(\"Model is \", model)\r\nprint(\"Model Stages\", model.stages)",
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "Using LogisticRegression model with Regularization Rate of 0.1.\nPipeline Created\nModel Trained\nModel is  PipelineModel_e5284bc61285\nModel Stages [StringIndexer_1ecf86c8d2ae, VectorAssembler_450ee37e6955, LogisticRegressionModel: uid = LogisticRegression_deb52c17940d, numClasses = 2, numFeatures = 2]"
+                    "text": "Using LogisticRegression model with Regularization Rate of 0.1.\nPipeline Created\nModel Trained\nModel is  PipelineModel_1adfacc01e7a\nModel Stages [StringIndexer_ee8506a28443, OneHotEncoderEstimator_cb5dbefb5cce, StringIndexer_38769cda5ab3, VectorAssembler_a3c2d358bd55, LogisticRegressionModel: uid = LogisticRegression_18837c9488f5, numClasses = 2, numFeatures = 17]",
+                    "output_type": "stream"
                 }
             ],
-            "execution_count": 7
+            "execution_count": 6
         },
         {
             "cell_type": "markdown",
@@ -113,12 +131,12 @@
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "Area under ROC: 0.7363559303440261\nArea Under PR: 0.39475773290351296\n+------+-----+----------+\n|income|label|prediction|\n+------+-----+----------+\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n+------+-----+----------+\nonly showing top 20 rows"
+                    "text": "Area under ROC: 0.7964496884726682\nArea Under PR: 0.5358180243123482\n+------+-----+----------+\n|income|label|prediction|\n+------+-----+----------+\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n| <=50K|  0.0|       1.0|\n|  >50K|  1.0|       1.0|\n+------+-----+----------+\nonly showing top 20 rows",
+                    "output_type": "stream"
                 }
             ],
-            "execution_count": 8
+            "execution_count": 7
         },
         {
             "cell_type": "markdown",
@@ -131,12 +149,12 @@
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "saved model to /spark_ml/AdultCensus.mml\nSuccessfully loaded from /spark_ml/AdultCensus.mml"
+                    "text": "saved model to /spark_ml/AdultCensus.mml\nSuccessfully loaded from /spark_ml/AdultCensus.mml",
+                    "output_type": "stream"
                 }
             ],
-            "execution_count": 9
+            "execution_count": 8
         },
         {
             "cell_type": "markdown",
@@ -149,12 +167,12 @@
             "metadata": {},
             "outputs": [
                 {
-                    "output_type": "stream",
                     "name": "stdout",
-                    "text": "persist the mleap bundle from local to hdfs"
+                    "text": "persist the mleap bundle from local to hdfs",
+                    "output_type": "stream"
                 }
             ],
-            "execution_count": 10
+            "execution_count": 9
         }
     ]
 }
\ No newline at end of file