Building Apache Spark Applications In Visual Studio

There are several applications named Spark. This post refers to Apache's in-memory streaming application suite named Spark. This is part of the Hadoop reference architecture. Generally speaking the documentation is oriented towards developers coding on a Linux laptop using sbt or Maven for CI support. I am working on a mixed team which prefers the Microsoft tooling (as do I). Our goal was to find a language that had built in Visual Studio and TFS support that would also deploy to HDFS using the Hadoop/Spark supported deployment tooling (Ambari and spark submit) with only minor environment reconfiguration.

Tooling:
Team Foundation Server 2013
Visual Studio 2013
Python Tools for Visual Studio (UnitTest, pip, python environment support, REPL integration, and MSBuild support for Python setup tools)
Spark 1.3
Python 2.7 (3.N, PyPy, and Anaconda are not tested with Spark yet)
CentOS
Windows 8.1
Hortonworks magical sandbox.
Powershell 3
GitHub for pulling Spark


Windows Setup
Install VS, PTVS, Python 2.7, IPython, nad GitHub to the default paths.
Install Spark in C:\Spark
Add a ton of environment variables (super important).

SPARK_HOME      C:\Spark

PYSPARK_HOME C:\Spark\Python

PY4J_HOME          C:\Spark\python\lib\py4j-0.8.2.1-src.zip

PYTHONPATH       C:\python27;C:\python27\scripts;c:\python27;c:\python27\scripts;%SPARK_HOME%;%PYSPARK_HOME%;%PY4J_HOME%

These are optional

PYTHON2      C:\python27\python

PYTHON3      C:\python3\python

ANACONDA C:\Users\ealdinger\AppData\Local\Continuum\Anaconda

GIT_HOME    <wherever you dump your files>

You can do this from Powershell or System - Advanced - Environment Variables

Testing Setup
Open Powershell

$py = $env:Path|select-string -pattern "c:\\python27"
$spark = $env:Path|select-string -pattern "c:\\spark"
$pyspark = $env:Path|select-string -pattern "c:\\spark\\python"
$py4j = $env:Path|select-string -pattern "C:\\Spark\\python\\lib\\py4j-0.8.2.1-src.zip"
$py -ne $null;$spark -ne $null;$pyspark -ne $null;$py4j -ne $null;

Start IPython and test that pyspark can be imported
Paste the lines below

from pyspark import SparkContext
logFile = "c:\spark\README.md"  # Should be some file on your system
sc = SparkContext("local", "SimpleApp")
logData = sc.textFile(logFile).cache()
numAs = logData.filter(lambda s: 'a' in s).count()
numBs = logData.filter(lambda s: 'b' in s).count()
print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"
print "Lines with a: %i, lines with b: %i" % (numAs, numBs)
print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"
sc.stop()

Look for

In [32]: print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"
_-_-_-_-_-_-_-_-_-_-_-_-_-_-
In [33]: print "Lines with a: %i, lines with b: %i" % (numAs, numBs)
Lines with a: 60, lines with b: 29
In [34]: print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"
_-_-_-_-_-_-_-_-_-_-_-_-_-_-

Start Visual Studio - Create a new Python Application project
Right click Search Paths in the solution.  Add PYTHONPATH to Search Path. You should see see spark, spark\python and py4j
Add a file to test with add

import re
for test_string in ['555-1212', 'ILL-EGAL']:
    if re.match(r'^\d{3}-\d{4}$', test_string):
        print test_string, 'is a valid US local phone number'
    else:
        print test_string, 'rejected'
print 'end of test'

Save and Start with Debugging
Add another file or change the first one.

from pyspark import SparkContext

logFile = "c:\spark\README.md"  # Should be some file on your system
sc = SparkContext("local", "Simple App")

logData = sc.textFile(logFile).cache()
numAs = logData.filter(lambda s: 'a' in s).count()
numBs = logData.filter(lambda s: 'b' in s).count()
print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"
print "Lines with a: %i, lines with b: %i" % (numAs, numBs)
print "_-_-_-_-_-_-_-_-_-_-_-_-_-_-"

sc.stop()

 Save and Start with Debugging. The script should run with a lot of output. The final output should be

Lines with a: 60, lines with b: 29

REF:
http://mund-consulting.com/Blog/using-ipython-and-visual-studio-with-apache-spark/

Some Generic Python Ideas

I am getting a chance to work with Python again. I am finding some features I was unaware of and like.

EXAMPLE 0

Iterating over slices of an array
>>> a = [1,2,3,4,5]
>>> a[::3]
[1, 4]
>>> a[::2]
[1, 3, 5]
>>> a[::5]
[1]
>>> a[::-1]
[5, 4, 3, 2, 1] //reverse order!
>>> b =["this", a, 234.0, 2012-12-12]
>>> b[::-1]
[1988, 234.0, [1, 2, 3, 4, 5], 'this']
>>> b[::2]
['this', 234.0]

This is a generator expression. Not sure why, as it seems like another general iterating thingie.
This is from the Pydocs functional programming page for version 2.

Iterating using a list comprehension
>>> words = "This is a set of fabulant vociferications".split()
>>> words
['This', 'is', 'a', 'set', 'of', 'fabulant', 'vociferications']
>>> letters = [len(word) for word in words]
>>> letters
[4, 2, 1, 3, 2, 8, 15]


EXAMPLE 1
>>> seq1 = 'abc'
>>> seq2 = ('1,2,3,4')
>>> [(x,y) for x in seq1 for y in seq2]

[('a', '1'), ('a', ','), ('a', '2'), ('a', ','), ('a', '3'), ('a', ','), ('a', '4'), ('b', '1'), ('b', ','), ('b', '2'), ('b', ','), ('b', '3'), ('b', ','), ('b', '4'), ('c', '1'), ('c', ','), ('c', '2'), ('c', ','), ('c', '3'), ('c', ','), ('c', '4')]

EXAMPLE 2
>>> seq1 = 'abc'
>>> seq2 = (1,2,3,4)
>>> seq3 = (5,6)
>>> [(x,y,z) for x in seq1 for y in seq2 for z in seq3]
[('a', 1, 5), ('a', 1, 6), ('a', 2, 5), ('a', 2, 6), ('a', 3, 5), ('a', 3, 6), ('a', 4, 5), ('a', 4, 6), ('b', 1, 5), ('
b', 1, 6), ('b', 2, 5), ('b', 2, 6), ('b', 3, 5), ('b', 3, 6), ('b', 4, 5), ('b', 4, 6), ('c', 1, 5), ('c', 1, 6), ('c',
 2, 5), ('c', 2, 6), ('c', 3, 5), ('c', 3, 6), ('c', 4, 5), ('c', 4, 6)]


Different ternary styles can be used.
These evaluate the same.
x=(2,3)[y==3]
x = 3 if (y==3) else 2
or you can branch logic
(func1 if y==2 else func2)(arg1,arg2)
or assignment
x = (class1 if y==2 else class2)(arg1,arg2)

EXAMPLE 4
While loop with nested until conditional logic common to python, using break statement to break a loop.
>>> while True:
...     response = input()
...     if int(response) %7==0:
...             break

EXAMPLE 5
Simple inheritance
class Reader(object):
    def __init__(self,fileName):
        self.fileName = FileName
        self.line

class reader2(Reader):
    def reader(self,fileName):
        try:
            f = open(fileName)
            s = f.readline()
            line = s.strip('a')
        except IOError as e:
            print "Error: {0}:{1}".format(e.errno,e.strerror)
        return line    

EXAMPLE 6
Shows how to default a pass parm, use a regex to detect if a pass parm is a URL, or raise an exception if a required parm is missing. TODO: add an argparse example.

import sys
import re
from urllib.request import urlopen
urlregx = re.compile('(https?:\/\/)([\da-z\.-]+)\.([a-z\.]{2,6})([\/\w \.-]*)*')

def fetch_words(_url):
    """Fetch the list of words from a url or a collection

    Args:
        url: a url to a utf-8 text

    Returns:
        A list of strings

    Tests:
        use 'http://sixty-north.com/c/t.txt' as url
    """
    with urlopen(_url) as story:
        story_words = []
        for line in story:
            line_words = line.decode('utf-8').split()
            for word in line_words:
                story_words.append(word)
    return story_words


def print_items(items):
    """print the items

        Args:
            An iterable series of non-object items
    """
    for item in items:
        print(item)


def main(_url = None):
    if _url == None:
        url = 'http://textfiles.com/anarchy/JOLLYROGER/001.jrc'
    else:
        url = _url
    words = fetch_words(url)
    print_items(words)

if __name__ == '__main__':
    _a = sys.argv
    try:
        x = urlregx.match(_a[1])
        if x == None:
            print(_a[1]+' is not a valid url')
            sys.exit()
        main(_a[1])
    except IndexError: 
        print("from the commandline include a parameter like 'http://textfiles.com/anarchy/JOLLYROGER/003.jrc'")
        sys.exit()

REFS
http://stackoverflow.com/questions/101268/hidden-features-of-python#112303
https://github.com/gregmalcolm/python_koans/wiki

http://ocw.mit.edu/courses/electrical-engineering-and-computer-science/6-189-a-gentle-introduction-to-programming-using-python-january-iap-2011/
https://docs.python.org/2/howto/functional.html

http://www.python-course.eu/lambda.php

FURTHER
How to do this in Python
    [BsonSerializer(typeof(DynamicMongoBsonSerializer))]
    public class DynamicMongoEntity : MongoEntityBase
    {
        public DynamicMongoEntity()
        {
            FieldsAsDynamic = new BsonDocument().ConvertBsonDocumentToDynamicObject();
        }

        /// <summary>
        /// BsonDocument representation of the FieldsAsDynamic property
        ///
        /// GETTER -> Will parse the BsonExtraElements document to json if it hasnt already
        /// SETTER -> Will set the local holder to the updated dynamic object and then deserize the dynamic json back into a BsonDocument for the BsonExtraElements
        /// </summary>
        public override BsonDocument CatchAll
        {
            get { return BsonSerializer.Deserialize<BsonDocument>(JsonConvert.SerializeObject(FieldsAsDynamic)); }
            set { FieldsAsDynamic = value.ConvertBsonDocumentToDynamicObject(); }
        }

        /// <summary>
        /// a c# dynamic representation of the BsonExtraElements document
        /// </summary>
        [BsonIgnore]
        [JsonIgnore]
        public dynamic FieldsAsDynamic { get; set; }

        /// <summary>
        /// This will find a dynamic mongo object in a give collection based on a key and value
        /// </summary>
        /// <param name="collection"></param>
        /// <param name="key"></param>
        /// <param name="value"></param>
        /// <returns></returns>
        public static DynamicMongoEntity FindByDynamicField(MongoCollection collection, string key, object value)
        {
            var dynamicQuery = Query.EQ(key, BsonValue.Create(value));

            var dynamicObject = collection.FindOneAs<DynamicMongoEntity>(dynamicQuery);

            return dynamicObject;
        }

        public static List<DynamicMongoEntity> FindAllByDynamicField(MongoCollection collection, string key, object value)
        {
            var dynamicQuery = Query.EQ(key, BsonValue.Create(value));

            var dynamicObjectList = collection.FindAs<DynamicMongoEntity>(dynamicQuery).ToList();

            return dynamicObjectList;
        }

        public static List<DynamicMongoEntity> FindAllByDynamicFields(MongoCollection collection, string[] keys, object[] values, Func<IEnumerable<IMongoQuery>, IMongoQuery> operatorFunc )
        {
            if (keys.Length != values.Length)
                throw new ArgumentException("Expected equal number of keys and values.");

            IEnumerable<IMongoQuery> queryExpressions = keys.Select((key, i) => Query.EQ(key, BsonValue.Create(values[i]))).ToList();

            var dynamicQuery = operatorFunc(queryExpressions);

            var dynamicObjectList = collection.FindAs<DynamicMongoEntity>(dynamicQuery).ToList();

            return dynamicObjectList;
        }

        public static IMongoQuery QueryDynamicField(string key, object value, Func<string, BsonValue, IMongoQuery> func)
        {
            return func(key, BsonValue.Create(value));
        }
    }

EXCEPTION HANDLING

#! python
# code page

import sys

"""module for exception testing"""

def convert(s):
        """converts to int"""
        x = -1.0
        try:
                x = int(s)
        except (ValueError,TypeError) as e:
                print("Conversion error: {}"\
                      .format(str(e)),file=sys.stderr)
        except:
                print("Life sucks then you die")
        return x

ITERATION EXAMPLES
iterable = ['Summer','Spring','Autumn','Winter']
iterator = iter(iterable)
for each in iterator:
...     next(iterator)Where next and iter are built ins and iterable and iterator are instance variables
 

Some Information on Data Visualization

Just a few links to help think about charting and color selection.

Notes
Color blindness need not drive every color palette choice if the graphic is also meaningful rendered without color contrast. Where possible provide either safe colors or strong complimentary visualization. If the objects are not discrete (orange and red palette in a heat map or stacked bar chart).
-me

1. Sequential schemes are suited to ordered data that progress from low to high. Lightness steps dominate the look of these schemes, with light colors for low data values to dark colors for high data values.

2. Diverging schemes put equal emphasis on mid-range critical values and extremes at both ends of the data range. The critical class or break in the middle of the legend is emphasized with light colors and low and high extremes are emphasized with dark colors that have contrasting hues.
Learn more »

3. Qualitative schemes do not imply magnitude differences between legend classes, and hues are used to create the primary visual differences between classes. Qualitative schemes are best suited to representing nominal or categorical data.
Learn more »

Brewer, Cynthia A. 1994. Color use guidelines for mapping and visualization. Chapter 7 (pp. 123-147) in Visualization in Modern Cartograph
Theory
http://blog.visual.ly/the-use-of-yellow-in-data-design/
http://www-psych.stanford.edu/~bt/diagrams/papers/diagramsstockholm04.pdf

Technologies
https://plot.ly/plot
http://blog.visual.ly/using-selections-in-d3-to-make-data-driven-visualizations/


Palettes
Look at the color palettes used by classic painters
http://visual.ly/10-artists-10-years-color-palettes
http://colorbrewer2.org/

Intro to Data Science - Threading Notes

In working on a simple tutorial I always tend to wander off on a tangent. This generally helps me engage the content more fully. In some cases it just confuses the heck out of me for a half hour. Looking at a threading tutorial in C# In A Nutshell I wanted to pass in instance variables to allow me to see the actual changes occurring as threads were being started. Seemed very straight forward to me, but I was getting Invalid Arguments errors. I found this finally "Thread methods needs (sic) to be a method with return type void and accepting no argument." S M Kamran on Stack Overflow. Now this got me wondering why is this such a bad idea it is disallowed.

    public class Asyncer
    {

        public static DateTime time1;
        public static long ticks1;
        public static DateTime time2 = System.DateTime.UtcNow;
        public static long ticks2 ;

        static void Main() {

            async();
            doSomeWork(" --A-- ", 12);
            doSomeWork(" --B-- ", 6);
            Console.ReadLine();
        }

        public static Thread async() {

            Thread threads = new Thread(doSomeWork(" --A-- ", 12));
            threads.Start();
       
        }
       
        public static void doSomeWork(string inPuter, Int32 inter) {
            for
                (Int32 cnt = 0; cnt <= inter; cnt++)

                //time1 = System.DateTime.UtcNow;
                //ticks1 = time1.Ticks;
                Console.Write(inPuter + Convert.ToString(ticks2) + Convert.ToString(ticks1));
        }
    }

Now this is not the point of what I was trying to learn, but a good question I wanted to write down for later.

Good things that I learned from this:

If you have a set of short lived threads you want to ensure to contain and sequence, use a thread pool.

Creating new threads is costly, so avoid it unless you want to force a thread to run in the foreground (taking precedence)  or actively synchronize long running threads. Another question for later is whether context sensitive threading requires creating the threads manually.

If you want to create threads you want to manage yourself, but do not need them to run in the foreground use  getter/setter threads.IsBackground = true; when newing up the thread.

The ThreadState property of a thread states whether the thread is foreground/background and running state.

That code executed in the Main method is a foreground thread. In the example below I execute two method calls and thread two more, where one thread is backgrounded and the other foregrounded.

This is the execution pattern for the code below on an i7 quad core laptop:
 --A-- 635590626227738088 BG 635590626227738090 FG 635590626227738089 BG 6355906
26228050713 FG 635590626228050712 --A-- 635590626228050711 --A-- 635590626228363
175 FG 635590626228363176 BG 635590626228363177 --A-- 635590626228675925 FG 6355
90626228675926 BG 635590626228675927 FG 635590626228988455 BG 635590626228988456
 --A-- 635590626228988454 FG 635590626229300719 --A-- 635590626229300717 BG 6355
90626229300719 FG 635590626229613232 BG 635590626229613234 --A-- 635590626229613
232 FG 635590626229925715 BG 635590626229925715 --A-- 635590626229925713 --A-- 6
35590626230238211 FG 635590626230238213 BG 635590626230238213 BG 635590626230550
780 --A-- 635590626230550778 FG 635590626230550779 FG 635590626230863286 BG 6355
90626230863287 --A-- 635590626230863285 FG 635590626231175797 BG 635590626231175
798 --A-- 635590626231175796 FG 635590626231488304 --A-- 635590626231488303 BG 6
35590626231488305 --B-- 635590626231800814 BG 635590626231800814 FG 635590626231
800815 FG 635590626232113326 BG 635590626232113327 --B-- 635590626232113325 FG 6
35590626232425841 --B-- 635590626232425840 BG 635590626232425842 FG 635590626232
738357 --B-- 635590626232738356 BG 635590626232738358 --B-- 635590626233050880 B
G 635590626233050880 FG 635590626233050881 BG 635590626233363393 FG 635590626233
363392 --B-- 635590626233363391 FG 635590626233675903 BG 635590626233675904 --B-
- 635590626233675902 BG 635590626233988419 FG 635590626233988418

I was surprised to that the background process is executed before the second execution of doSomeWork is ever executed. It makes sense that all executions of the first instance of doSomeWork must complete before the second, and that there is processor available for the BG process while B is waiting. Not as surprising is that the Console also interjects some syncronicity issues in returning the results. We see the last few executions occured in this predictable order
--B-- 635590626233363391
--B-- 635590626233675902
FG 635590626233675903
BG 635590626233675904
FG 635590626233988418

BG 635590626233988419

But were written in this WTF order:
--B-- 635590626233363391
FG 635590626233675903
BG 635590626233675904
--B-- 635590626233675902
BG 635590626233988419
FG 635590626233988418

 
using System;
using System.Threading;

namespace AsyncProject
{
    public class Asyncer
    {
        public static DateTime time1;
        public static DateTime time2 = System.DateTime.UtcNow;
        public static long ticks1;
        public static long ticks2;

        static void Main() {

            //execute a background thread
            Thread threads = new Thread(doSomeWorkFC);
            threads.IsBackground = true;
            threads.Start();

            //execute a foreground thread
            Thread threader = new Thread(doSomeWorkBG);
            threader.IsBackground = false; //default value setting for example clarity
            threader.Start();

            //execute some methods in foreground w/o explicitly creating a thread
            doSomeWork(" --A-- ", 12);
            doSomeWork(" --B-- ", 6);

            //hold open the console
            Console.ReadLine();
        }
        public static void doSomeWork(string inPuter, Int32 inter)
        {
            for
                (Int32 cnt = 0; cnt <= inter; cnt++)
            {
                time1 = System.DateTime.UtcNow;
                ticks1 = time1.Ticks;
                Console.Write(inPuter + Convert.ToString(ticks1));
                Thread.Sleep(20);
            }
        }

        public static void doSomeWorkFC()
        {
            for
                (Int32 cnt = 0; cnt <= 20; cnt++)
            {
                time1 = System.DateTime.UtcNow;
                ticks1 = time1.Ticks+1;
                Console.Write(" FG " + Convert.ToString(ticks1));
                Thread.Sleep(20);
            }
        }

        public static void doSomeWorkBG()
        {
            for
                (Int32 cnt = 0; cnt <= 20; cnt++)
            {
                time1 = System.DateTime.UtcNow;
                ticks1 = time1.Ticks+2;
                Console.Write(" BG " + Convert.ToString(ticks1));
                Thread.Sleep(20);
            }
        }
    }
}



REF:
http://stackoverflow.com/questions/230003/thread-vs-threadpool
http://stackoverflow.com/questions/5155979/c-sharp-thread-method

SQL To Monitor Table Load Progress

DECLARE @1 TABLE (Startcnt1 BIGINT, Startcnt2 BIGINT,Startcnt3 BIGINT, Secondcnt1 BIGINT, Secondcnt2 BIGINT, Secondcnt3 BIGINT)


INSERT INTO @1
        ( Startcnt1 ,
          Startcnt2 ,
          Startcnt3
        )
SELECT
(SELECT COUNT(1) FROM Table1),
(SELECT COUNT(1) FROM Table2),
(SELECT COUNT(1) FROM Table3)

WAITFOR DELAY '00:01'

UPDATE @1
SET Secondcnt1 = (SELECT COUNT(1) FROM Table1)

UPDATE @1
SET Secondcnt2 = (SELECT COUNT(1) FROM Table2)

UPDATE @1
SET Secondcnt3 = (SELECT COUNT(1) FROM Table3)

SELECT Startcnt1 ,
       Secondcnt1 ,
       Secondcnt1 -Startcnt1 ChangeCnt1,
       Startcnt2 ,
       Secondcnt2 ,
Secondcnt2 - Startcnt2  ChangeCnt2,
       Startcnt3 ,
       Secondcnt3 ,
Secondcnt3 - Startcnt3  ChangeCnt3
FROM @1

Intro To Data Science - NoSQL Notes

Time to learn a thing or two about the world outside of RDBMS and ACID. one of the hardest things to do is to walk away from something I know intuitively and embrace something foreign.

Mongo -
Document based. BSON storage. You can link document collections or embed documents. Adding attributes to the schema is extending the schema. As you extend you add an index, or extend an existing one (if the index needs to be covering rather than an access path to a record). One presentation stated "each attribute needs an index, each time you extend you add an index" I am not sure all attributes need an index.

KVP - key value pair association is simple and effective, but lacks the ability to use the complex relation and retrieval syntax associated with ANSI 92 SQL. A pure key/value store is completely agnostic toward the data stored within it, including data types. Simplicity allows for quick read and write. Not so easy to access using fuzzy logic or grouping/composite queries.

Secondary indexes are a feature that (not to be confused with clustered indexes or primary keys) allow tagging of KVP for alternate access paths. An example is a record with a secondary index

User.getIndexes().getIndex(StringBinIndex.named("ST")).add("GA");

User.getIndexes().getIndex(LongIntIndex.named("ALTID")).add(972121001L); 

Buckets - used to define a virtual keyspace for storing Riak objects. They enable you to define non-default configurations over that keyspace concerning replication properties and other parameters.

Good article on LinkedIn's approach. Simple and surficial, it provides basic context for some key considerations I am listing here
http://www.slideshare.net/amywtang/espresso-20952131
https://gigaom.com/2014/11/26/linkedin-explains-its-complex-gobblin-big-data-framework/

Riak examples are simple and easy to understand. Hurt my head at first just because it is so different.
http://docs.basho.com/riak/latest/dev/using/2i/


MongoDB Array Updates

db.test.remove({"item" : "ABC1"});

db.test.insert(
{
    "item" : "ABC1",
    "details" : {
        "model" : "14Q3",
        "manufacturer" : "XYZ Company"
    },
    "stock" : [
        {
            "size" : "S",
            "qty" : 25
        },
        {
            "size" : "M",
            "qty" : 50
        }
    ],
    "category" : "clothing"
});

//do not do this. it replaces the document
//db.test.update(
//{$and:[ {"item" : "ABC1"}, {"stock.size":"M"} ]},
//{"qty":45}
//);

db.test.update(
{$and:[ {"item" : "ABC1"}, {"stock.size":"M"} ]},
{$set: {"stock.$.qty":45}}
);

db.test.find({$and:[ {"item" : "ABC1"}, {"stock.size":"M"} ]});
 

Conditionally Running SQL Based on Server

In data operations we often have to limit access to personally identifiable information (PII) in environments with fewer safe guards. So in production we may import personal identifiers like credit card numbers or government issued ids (social security numbers in the US). In a development environment we may want to treat that data with a mask on importing it. This can require two different import processes if written incorrectly, meaning that which is tested in dev or QA is not that which is promoted to production. Conditional logic can be based on server name, if that name is a consistent alias for the host managing the distributed transaction coordination. Here is a template.

USE <DATABASE>


IF (SELECT @@SERVERNAME) IN ('Server1\Instance', 'Server2')
BEGIN

    BEGIN TRY
        --insert your production code here

        --no changes past here
        COMMIT
    END TRY

    BEGIN CATCH
        ROLLBACK
        PRINT ERROR_MESSAGE()
    END CATCH
END


IF (SELECT @@SERVERNAME) NOT IN ('Server1\Instance', 'Server2')
BEGIN

    BEGIN TRY
        --insert your SPI masking code here

        --no changes past here
        COMMIT
    END TRY

    BEGIN CATCH
        ROLLBACK
        PRINT ERROR_MESSAGE()
    END CATCH
END