diff --git a/remotespark/default_config.json b/remotespark/default_config.json
index 3be83bf..ea4adaa 100644
--- a/remotespark/default_config.json
+++ b/remotespark/default_config.json
@@ -47,5 +47,10 @@
 
   "fatal_error_suggestion": "The code failed because of a fatal error:\n\t{}.\n\nSome things to try:\na) Make sure Spark has enough available resources for Jupyter to create a Spark context.\nb) Contact your Jupyter administrator to make sure the Spark magics library is configured correctly.\nc) Restart the kernel.",
 
-  "ignore_ssl_errors": false
+  "ignore_ssl_errors": false,
+
+  "session_configs": {
+    "driverMemory": "1000M",
+    "executorCores": 2
+  }
 }
diff --git a/remotespark/livyclientlib/clientmanagerstateserializer.py b/remotespark/livyclientlib/clientmanagerstateserializer.py
index 8a04bce..f4615ea 100644
--- a/remotespark/livyclientlib/clientmanagerstateserializer.py
+++ b/remotespark/livyclientlib/clientmanagerstateserializer.py
@@ -38,11 +38,11 @@ class ClientManagerStateSerializer(object):
                 name = client["name"]
                 session_id = client["id"]
                 sql_context_created = client["sqlcontext"]
-                language = client["language"]
+                kind = client["kind"].lower()
                 connection_string = client["connectionstring"]
 
                 session = self._client_factory.create_session(
-                    language, connection_string, session_id, sql_context_created)
+                    connection_string, session_id, sql_context_created, {"kind": kind})
 
                 # Do not start session automatically. Just create it but skip is not existent.
                 try:
@@ -50,7 +50,7 @@ class ClientManagerStateSerializer(object):
                     status = session.status
                     if not session.is_final_status(status):
                         self.logger.debug("Adding session {}".format(session_id))
-                        client_obj = self._client_factory.build_client(language, session)
+                        client_obj = self._client_factory.build_client(session)
                         clients_to_return.append((name, client_obj))
                     else:
                         self.logger.error("Skipping serialized session '{}' because session was in status {}."
diff --git a/remotespark/livyclientlib/livyclient.py b/remotespark/livyclientlib/livyclient.py
index 6632b2a..f216c44 100644
--- a/remotespark/livyclientlib/livyclient.py
+++ b/remotespark/livyclientlib/livyclient.py
@@ -34,8 +34,8 @@ class LivyClient(object):
         self._session.delete()
 
     @property
-    def language(self):
-        return self._session.language
+    def kind(self):
+        return self._session.kind
 
     @property
     def session_id(self):
diff --git a/remotespark/livyclientlib/livyclientfactory.py b/remotespark/livyclientlib/livyclientfactory.py
index 7599ddd..f7b2ae3 100644
--- a/remotespark/livyclientlib/livyclientfactory.py
+++ b/remotespark/livyclientlib/livyclientfactory.py
@@ -17,23 +17,24 @@ class LivyClientFactory(object):
         self.logger = Log("LivyClientFactory")
         self.max_results = 2500
 
-    def build_client(self, language, session):
+    def build_client(self, session):
         assert session is not None
+        kind = session.kind
 
-        if language == Constants.lang_python:
+        if kind == Constants.session_kind_pyspark:
             return PandasPysparkLivyClient(session, self.max_results)
-        elif language == Constants.lang_scala:
+        elif kind == Constants.session_kind_spark:
             return PandasScalaLivyClient(session, self.max_results)
         else:
-            raise ValueError("Language '{}' is not supported.".format(language))
+            raise ValueError("Kind '{}' is not supported.".format(kind))
 
     @staticmethod
-    def create_session(language, connection_string, session_id="-1", sql_created=False):
+    def create_session(connection_string, properties, session_id="-1", sql_created=False):
         cso = get_connection_string_elements(connection_string)
 
         retry_policy = LinearRetryPolicy(seconds_to_sleep=5, max_retries=5)
         http_client = LivyReliableHttpClient(cso.url, cso.username, cso.password, retry_policy)
 
-        session = LivySession(http_client, language, session_id, sql_created)
+        session = LivySession(http_client, session_id, sql_created, properties)
 
         return session
diff --git a/remotespark/livyclientlib/livysession.py b/remotespark/livyclientlib/livysession.py
index 247ab84..7c8efaf 100644
--- a/remotespark/livyclientlib/livysession.py
+++ b/remotespark/livyclientlib/livysession.py
@@ -7,7 +7,6 @@ from time import sleep, time
 import remotespark.utils.configuration as conf
 from remotespark.utils.constants import Constants
 from remotespark.utils.log import Log
-from remotespark.utils.utils import get_instance_id
 from .livyclienttimeouterror import LivyClientTimeoutError
 from .livyunexpectedstatuserror import LivyUnexpectedStatusError
 from .livysessionstate import LivySessionState
@@ -15,9 +14,12 @@ from .livysessionstate import LivySessionState
 
 class LivySession(object):
     """Session that is livy specific."""
-    # TODO(aggftw): make threadsafe
 
-    def __init__(self, http_client, language, session_id, sql_created):
+    def __init__(self, http_client, session_id, sql_created, properties):
+        assert "kind" in properties.keys()
+        kind = properties["kind"]
+        self.properties = properties
+
         status_sleep_seconds = conf.status_sleep_seconds()
         statement_sleep_seconds = conf.statement_sleep_seconds()
         create_sql_context_timeout_seconds = conf.create_sql_context_timeout_seconds()
@@ -30,10 +32,10 @@ class LivySession(object):
 
         self.logger = Log("LivySession")
 
-        language = language.lower()
-        if language not in Constants.lang_supported:
-            raise ValueError("Session of language '{}' not supported. Session must be of languages {}."
-                             .format(language, ", ".join(Constants.lang_supported)))
+        kind = kind.lower()
+        if kind not in Constants.session_kinds_supported:
+            raise ValueError("Session of kind '{}' not supported. Session must be of kinds {}."
+                             .format(kind, ", ".join(Constants.session_kinds_supported)))
 
         if session_id == "-1":
             self._status = Constants.not_started_session_status
@@ -41,45 +43,44 @@ class LivySession(object):
         else:
             self._status = Constants.busy_session_status
 
+        self._logs = ""
         self._http_client = http_client
         self._status_sleep_seconds = status_sleep_seconds
         self._statement_sleep_seconds = statement_sleep_seconds
         self._create_sql_context_timeout_seconds = create_sql_context_timeout_seconds
 
         self._state = LivySessionState(session_id, http_client.connection_string,
-                                       language, sql_created)
+                                       kind, sql_created)
 
     def get_state(self):
         return self._state
 
     def start(self):
         """Start the session against actual livy server."""
-        # TODO(aggftw): do a pass to make all contracts variables; i.e. not peppered in code
-        self.logger.debug("Starting '{}' session.".format(self.language))
+        self.logger.debug("Starting '{}' session.".format(self.kind))
 
-        app_name = "remotesparkmagics_{}".format(get_instance_id())
-        r = self._http_client.post("/sessions", [201], {"kind": self._get_livy_kind(), "name": app_name})
+        r = self._http_client.post("/sessions", [201], self.properties)
         self._state.session_id = str(r.json()["id"])
         self._status = str(r.json()["state"])
 
-        self.logger.debug("Session '{}' started.".format(self.language))
+        self.logger.debug("Session '{}' started.".format(self.kind))
 
     def create_sql_context(self):
         """Create a sqlContext object on the session. Object will be accessible via variable 'sqlContext'."""
         if self.started_sql_context:
             return
 
-        self.logger.debug("Starting '{}' sql and hive session.".format(self.language))
+        self.logger.debug("Starting '{}' sql and hive session.".format(self.kind))
 
-        self._create_context("sql")
-        self._create_context("hive")
+        self._create_context(Constants.context_name_sql)
+        self._create_context(Constants.context_name_hive)
 
         self._state.sql_context_created = True
 
     def _create_context(self, context_type):
-        if context_type == "sql":
+        if context_type == Constants.context_name_sql:
             command = self._get_sql_context_creation_command()
-        elif context_type == "hive":
+        elif context_type == Constants.context_name_hive:
             command = self._get_hive_context_creation_command()
         else:
             raise ValueError("Cannot create context of type {}.".format(context_type))
@@ -87,7 +88,7 @@ class LivySession(object):
         try:
             self.wait_for_idle(self._create_sql_context_timeout_seconds)
             self.execute(command)
-            self.logger.debug("Started '{}' {} session.".format(self.language, context_type))
+            self.logger.debug("Started '{}' {} session.".format(self.kind, context_type))
         except LivyClientTimeoutError:
             raise LivyClientTimeoutError("Failed to create the {} context in time. Timed out after {} seconds."
                                          .format(context_type, self._create_sql_context_timeout_seconds))
@@ -101,8 +102,8 @@ class LivySession(object):
         return self._state.sql_context_created
 
     @property
-    def language(self):
-        return self._state.language
+    def kind(self):
+        return self._state.kind
 
     def refresh_status(self):
         (status, logs) = self._get_latest_status_and_logs()
@@ -124,7 +125,6 @@ class LivySession(object):
         return status in Constants.final_status
     
     def execute(self, commands):
-        """Executes commands in session."""
         code = textwrap.dedent(commands)
 
         data = {"code": code}
@@ -134,7 +134,6 @@ class LivySession(object):
         return self._get_statement_output(statement_id)
 
     def delete(self):
-        """Deletes the session and releases any resources."""
         self.logger.debug("Deleting session '{}'".format(self.id))
 
         if self._status != Constants.not_started_session_status and self._status != Constants.dead_session_status:
@@ -147,7 +146,11 @@ class LivySession(object):
 
     def wait_for_idle(self, seconds_to_wait):
         """Wait for session to go to idle status. Sleep meanwhile. Calls done every status_sleep_seconds as
-        indicated by the constructor."""
+        indicated by the constructor.
+
+        Parameters:
+            seconds_to_wait : number of seconds to wait before giving up.
+        """
 
         self.refresh_status()
         current_status = self._status
@@ -187,11 +190,11 @@ class LivySession(object):
                              .format(self.id, len(filtered_sessions)))
             
         session = filtered_sessions[0]
-        return (session['state'], session['log'])
+        return session['state'], session['log']
     
     def _get_statement_output(self, statement_id):
         statement_running = True
-        output = ""
+        out = ""
         while statement_running:
             r = self._http_client.get(self._statements_url(), [200])
             statement = [i for i in r.json()["statements"] if i["id"] == statement_id][0]
@@ -208,36 +211,31 @@ class LivySession(object):
                 if statement_output["status"] == "ok":
                     out = (True, statement_output["data"]["text/plain"])
                 elif statement_output["status"] == "error":
-                    out = (False, statement_output["evalue"] + "\n" + \
-                                  "".join(statement_output["traceback"]))
-        return out
+                    out = (False, statement_output["evalue"] + "\n" +
+                           "".join(statement_output["traceback"]))
+                else:
+                    raise ValueError("Unknown output status: '{}'".format(statement_output["status"]))
 
-    def _get_livy_kind(self):
-        if self.language == Constants.lang_scala:
-            return Constants.session_kind_spark
-        elif self.language == Constants.lang_python:
-            return Constants.session_kind_pyspark
-        else:
-            raise ValueError("Cannot get session kind for {}.".format(self.language))
+        return out
 
     def _get_sql_context_creation_command(self):
-        if self.language == Constants.lang_scala:
+        if self.kind == Constants.session_kind_spark:
             sql_context_command = "val sqlContext = new org.apache.spark.sql.SQLContext(sc)\n" \
                                   "import sqlContext.implicits._"
-        elif self.language == Constants.lang_python:
+        elif self.kind == Constants.session_kind_pyspark:
             sql_context_command = "from pyspark.sql import SQLContext\nfrom pyspark.sql.types import *\n" \
                                   "sqlContext = SQLContext(sc)"
         else:
-            raise ValueError("Do not know how to create sqlContext in session of language {}.".format(self.language))
+            raise ValueError("Do not know how to create sqlContext in session of kind {}.".format(self.kind))
 
         return sql_context_command
 
     def _get_hive_context_creation_command(self):
-        if self.language == Constants.lang_scala:
+        if self.kind == Constants.session_kind_spark:
             hive_context_command = "val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)"
-        elif self.language == Constants.lang_python:
+        elif self.kind == Constants.session_kind_pyspark:
             hive_context_command = "from pyspark.sql import HiveContext\nhiveContext = HiveContext(sc)"
         else:
-            raise ValueError("Do not know how to create hiveContext in session of language {}.".format(self.language))
+            raise ValueError("Do not know how to create hiveContext in session of kind {}.".format(self.kind))
 
         return hive_context_command
diff --git a/remotespark/livyclientlib/livysessionstate.py b/remotespark/livyclientlib/livysessionstate.py
index 66654f1..c86dc97 100644
--- a/remotespark/livyclientlib/livysessionstate.py
+++ b/remotespark/livyclientlib/livysessionstate.py
@@ -3,9 +3,9 @@
 
 
 class LivySessionState(object):
-    def __init__(self, session_id, connection_string, language, sql_context_created, version="0.0.0"):
+    def __init__(self, session_id, connection_string, kind, sql_context_created, version="0.0.0"):
         self._session_id = session_id
-        self._language = language
+        self._kind = kind
         self._sql_context_created = sql_context_created
         self._version = version
         self._connection_string = connection_string
@@ -19,8 +19,8 @@ class LivySessionState(object):
         self._session_id = value
 
     @property
-    def language(self):
-        return self._language
+    def kind(self):
+        return self._kind
 
     @property
     def sql_context_created(self):
@@ -39,5 +39,5 @@ class LivySessionState(object):
         return self._connection_string
 
     def to_dict(self):
-        return {"id": self.session_id, "language": self.language, "sqlcontext": self.sql_context_created,
+        return {"id": self.session_id, "kind": self.kind, "sqlcontext": self.sql_context_created,
                 "version": self.version, "connectionstring": self.connection_string}
diff --git a/remotespark/livyclientlib/sparkcontroller.py b/remotespark/livyclientlib/sparkcontroller.py
index b3aa9d0..b5ef94a 100644
--- a/remotespark/livyclientlib/sparkcontroller.py
+++ b/remotespark/livyclientlib/sparkcontroller.py
@@ -38,14 +38,14 @@ class SparkController(object):
     def delete_session(self, name):
         self.client_manager.delete_client(name)
 
-    def add_session(self, name, language, connection_string, skip_if_exists):
+    def add_session(self, name, connection_string, skip_if_exists, properties):
         if skip_if_exists and (name in self.client_manager.get_sessions_list()):
             self.logger.debug("Skipping {} because it already exists in list of sessions.".format(name))
             return
 
-        session = self.client_factory.create_session(language, connection_string, "-1", False)
+        session = self.client_factory.create_session(connection_string, properties, "-1", False)
         session.start()
-        livy_client = self.client_factory.build_client(language, session)
+        livy_client = self.client_factory.build_client(session)
         self.client_manager.add_client(name, livy_client)
 
     def get_client_keys(self):
diff --git a/remotespark/remotesparkmagics.py b/remotespark/remotesparkmagics.py
index ee17f21..821eef8 100644
--- a/remotespark/remotesparkmagics.py
+++ b/remotespark/remotesparkmagics.py
@@ -7,6 +7,8 @@ Provides the %spark magic."""
 from __future__ import print_function
 from IPython.core.magic import Magics, magics_class, line_cell_magic, needs_local_scope
 from IPython.core.magic_arguments import argument, magic_arguments, parse_argstring
+import json
+import copy
 
 import remotespark.utils.configuration as conf
 from remotespark.utils.constants import Constants
@@ -41,6 +43,8 @@ class RemoteSparkMagics(Magics):
         except KeyError:
             self.logger.error("Could not read env vars for serialization.")
 
+        self.properties = conf.session_configs()
+
         self.logger.debug("Initialized spark magics.")
 
     @magic_arguments()
@@ -57,7 +61,7 @@ class RemoteSparkMagics(Magics):
     @argument("command", type=str, default=[""], nargs="*", help="Commands to execute.")
     @needs_local_scope
     @line_cell_magic
-    def spark(self, line, cell="", local_ns={}):
+    def spark(self, line, cell="", local_ns=None):
         """Magic to execute spark remotely.
 
            This magic allows you to create a Livy Scala or Python session against a Livy endpoint. Every session can
@@ -68,21 +72,28 @@ class RemoteSparkMagics(Magics):
            Subcommands
            -----------
            info
-               Display the mode and available Livy sessions.
+               Display the available Livy sessions and other configurations for sessions.
            add
                Add a Livy session. First argument is the name of the session, second argument
                is the language, and third argument is the connection string of the Livy endpoint.
                A fourth argument specifying if session creation can be skipped if it already exists is optional:
                "skip" or empty.
-               e.g. `%%spark add test python url=https://sparkcluster.example.net/livy;username=admin;password=MyPassword skip`
+               e.g. `%%spark add test python url=https://sparkcluster.net/livy;username=u;password=p skip`
                or
-               e.g. `%%spark add test python url=https://sparkcluster.example.net/livy;username=admin;password=MyPassword`
+               e.g. `%%spark add test python url=https://sparkcluster.net/livy;username=u;password=p`
+           config
+               Override the livy session properties sent to Livy on session creation. All session creations will
+               contain these config settings from then on.
+               Expected value is a JSON key-value string to be sent as part of the Request Body for the POST /sessions
+               endpoint in Livy.
+               e.g. `%%spark config {"driverMemory":"1000M", "executorCores":4}`
            run
                Run Spark code against a session.
                e.g. `%%spark -e testsession` will execute the cell code against the testsession previously created
                e.g. `%%spark -e testsession -c sql` will execute the SQL code against the testsession previously created
-               e.g. `%%spark -e testsession -c sql -o my_var` will execute the SQL code against the testsession previously
-                        created and store the pandas dataframe created in the my_var variable in the Python environment
+               e.g. `%%spark -e testsession -c sql -o my_var` will execute the SQL code against the testsession
+                        previously created and store the pandas dataframe created in the my_var variable in the
+                        Python environment.
            delete
                Delete a Livy session. Argument is the name of the session to be deleted.
                e.g. `%%spark delete defaultlivy`
@@ -99,18 +110,29 @@ class RemoteSparkMagics(Magics):
         # info
         if subcommand == "info":
             self._print_info()
+        # config
+        elif subcommand == "config":
+            # Would normally do " ".join(args.command[1:]) but parse_argstring removes quotes...
+            rest_of_line = user_input[7:]
+            self.properties = json.loads(rest_of_line)
         # add
         elif subcommand == "add":
             if len(args.command) != 4 and len(args.command) != 5:
                 raise ValueError("Subcommand 'add' requires three or four arguments. {}".format(usage))
+
             name = args.command[1].lower()
-            language = args.command[2]
+            language = args.command[2].lower()
             connection_string = args.command[3]
+
             if len(args.command) == 5:
                 skip = args.command[4].lower() == "skip"
             else:
                 skip = False
-            self.spark_controller.add_session(name, language, connection_string, skip)
+
+            properties = copy.deepcopy(self.properties)
+            properties["kind"] = self._get_livy_kind(language)
+
+            self.spark_controller.add_session(name, connection_string, skip, properties)
         # delete
         elif subcommand == "delete":
             if len(args.command) != 2:
@@ -151,7 +173,21 @@ class RemoteSparkMagics(Magics):
             return None
 
     def _print_info(self):
-        print("Info for running Spark:\n\t{}\n".format(self.spark_controller.get_client_keys()))
+        print("""Info for running Spark:
+    Sessions:
+        {}
+    Session configs:
+        {}
+""".format(self.spark_controller.get_client_keys(), self.properties))
+
+    @staticmethod
+    def _get_livy_kind(language):
+        if language == Constants.lang_scala:
+            return Constants.session_kind_spark
+        elif language == Constants.lang_python:
+            return Constants.session_kind_pyspark
+        else:
+            raise ValueError("Cannot get session kind for {}.".format(language))
 
         
 def load_ipython_extension(ip):
diff --git a/remotespark/utils/configuration.py b/remotespark/utils/configuration.py
index cdef740..df09e52 100644
--- a/remotespark/utils/configuration.py
+++ b/remotespark/utils/configuration.py
@@ -68,6 +68,10 @@ def _override(f):
 # value of that configuration if there is any such configuration. Otherwise,
 # these functions return the default values described in their bodies.
 
+@_override
+def session_configs():
+    return {}
+
 @_override
 def serialize():
     return False
diff --git a/remotespark/utils/constants.py b/remotespark/utils/constants.py
index 197bd65..6d6a1cc 100644
--- a/remotespark/utils/constants.py
+++ b/remotespark/utils/constants.py
@@ -7,6 +7,7 @@ class Constants:
 
     session_kind_spark = "spark"
     session_kind_pyspark = "pyspark"
+    session_kinds_supported = [session_kind_spark, session_kind_pyspark]
 
     context_name_spark = "spark"
     context_name_sql = "sql"

Allow user to specify memory/cores/etc for every session
diff --git a/tests/test_clientmanagerstateserializer.py b/tests/test_clientmanagerstateserializer.py
index 4310058..4ed8ed7 100644
--- a/tests/test_clientmanagerstateserializer.py
+++ b/tests/test_clientmanagerstateserializer.py
@@ -27,7 +27,7 @@ def test_deserialize_not_emtpy():
       "name": "py",
       "id": "1",
       "sqlcontext": true,
-      "language": "python",
+      "kind": "pyspark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     },
@@ -35,7 +35,7 @@ def test_deserialize_not_emtpy():
       "name": "sc",
       "id": "2",
       "sqlcontext": false,
-      "language": "scala",
+      "kind": "spark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     }
@@ -50,17 +50,15 @@ def test_deserialize_not_emtpy():
 
     (name, client) = deserialized[0]
     assert name == "py"
-    client_factory.create_session.assert_any_call("python",
-                                                  "url=https://mysite.com/livy;username=user;password=pass",
-                                                  "1", True)
-    client_factory.build_client.assert_any_call("python", session)
+    client_factory.create_session.assert_any_call("url=https://mysite.com/livy;username=user;password=pass",
+                                                  "1", True, {"kind":"pyspark"})
+    client_factory.build_client.assert_any_call(session)
 
     (name, client) = deserialized[1]
     assert name == "sc"
-    client_factory.create_session.assert_any_call("scala",
-                                                  "url=https://mysite.com/livy;username=user;password=pass",
-                                                  "2", False)
-    client_factory.build_client.assert_any_call("scala", session)
+    client_factory.create_session.assert_any_call("url=https://mysite.com/livy;username=user;password=pass",
+                                                  "2", False, {"kind":"spark"})
+    client_factory.build_client.assert_any_call(session)
 
 
 def test_deserialize_not_emtpy_but_dead():
@@ -75,7 +73,7 @@ def test_deserialize_not_emtpy_but_dead():
       "name": "py",
       "id": "1",
       "sqlcontext": true,
-      "language": "python",
+      "kind": "pyspark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     },
@@ -83,7 +81,7 @@ def test_deserialize_not_emtpy_but_dead():
       "name": "sc",
       "id": "2",
       "sqlcontext": false,
-      "language": "scala",
+      "kind": "spark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     }
@@ -113,7 +111,7 @@ def test_deserialize_not_emtpy_but_error():
       "name": "py",
       "id": "1",
       "sqlcontext": true,
-      "language": "python",
+      "kind": "pyspark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     },
@@ -121,7 +119,7 @@ def test_deserialize_not_emtpy_but_error():
       "name": "sc",
       "id": "2",
       "sqlcontext": false,
-      "language": "scala",
+      "kind": "spark",
       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
       "version": "0.0.0"
     }
@@ -155,11 +153,11 @@ def test_serialize_not_empty():
     client_factory = MagicMock()
     reader_writer = MagicMock()
     client1 = MagicMock()
-    client1.serialize.return_value = {"id": "1", "sqlcontext": True, "language": "python",
+    client1.serialize.return_value = {"id": "1", "sqlcontext": True, "kind": "pyspark",
                                       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
                                       "version": "0.0.0"}
     client2 = MagicMock()
-    client2.serialize.return_value = {"id": "2", "sqlcontext": False, "language": "scala",
+    client2.serialize.return_value = {"id": "2", "sqlcontext": False, "kind": "spark",
                                       "connectionstring": "url=https://mysite.com/livy;username=user;password=pass",
                                       "version": "0.0.0"}
     serializer = ClientManagerStateSerializer(client_factory, reader_writer)
@@ -169,9 +167,9 @@ def test_serialize_not_empty():
 
     # Verify write was called with following string
     expected_str = '{"clients": [{"name": "py", "connectionstring": "url=https://mysite.com/livy;username=user;p' \
-                   'assword=pass", "version": "0.0.0", "language": "python", "sqlcontext": true, "id": "1"}, {"n' \
+                   'assword=pass", "version": "0.0.0", "kind": "pyspark", "sqlcontext": true, "id": "1"}, {"n' \
                    'ame": "sc", "connectionstring": "url=https://mysite.com/livy;username=user;password=pass", "ve' \
-                   'rsion": "0.0.0", "language": "scala", "sqlcontext": false, "id": "2"}]}'
+                   'rsion": "0.0.0", "kind": "spark", "sqlcontext": false, "id": "2"}]}'
     expected_dict = json.loads(expected_str)
     call_list = reader_writer.overwrite_with_line.call_args_list
     assert len(call_list) == 1
diff --git a/tests/test_livyclient.py b/tests/test_livyclient.py
index a47e023..57870fa 100644
--- a/tests/test_livyclient.py
+++ b/tests/test_livyclient.py
@@ -3,6 +3,7 @@
 from remotespark.livyclientlib.livyclient import LivyClient
 from remotespark.livyclientlib.livysessionstate import LivySessionState
 from remotespark.utils.utils import get_connection_string
+from remotespark.utils.constants import Constants
 
 
 def test_create_sql_context_automatically():
@@ -55,7 +56,7 @@ def test_serialize():
     connection_string = get_connection_string(url, username, password)
     http_client = MagicMock()
     http_client.connection_string = connection_string
-    kind = "scala"
+    kind = Constants.session_kind_spark
     session_id = "-1"
     sql_created = False
     session = MagicMock()
@@ -67,7 +68,7 @@ def test_serialize():
 
     assert serialized["connectionstring"] == connection_string
     assert serialized["id"] == "-1"
-    assert serialized["language"] == kind
+    assert serialized["kind"] == kind
     assert serialized["sqlcontext"] == sql_created
     assert serialized["version"] == "0.0.0"
     assert len(serialized.keys()) == 5
@@ -82,16 +83,16 @@ def test_close_session():
     mock_spark_session.delete.assert_called_once_with()
 
 
-def test_language():
-    lang = "python"
+def test_kind():
+    kind = "pyspark"
     mock_spark_session = MagicMock()
-    language_mock = PropertyMock(return_value=lang)
-    type(mock_spark_session).language = language_mock
+    language_mock = PropertyMock(return_value=kind)
+    type(mock_spark_session).kind = language_mock
     client = LivyClient(mock_spark_session)
 
-    l = client.language
+    l = client.kind
 
-    assert l == lang
+    assert l == kind
 
 
 def test_session_id():
diff --git a/tests/test_livyclientfactory.py b/tests/test_livyclientfactory.py
index 1e2e4f2..643b103 100644
--- a/tests/test_livyclientfactory.py
+++ b/tests/test_livyclientfactory.py
@@ -2,6 +2,8 @@ from mock import MagicMock
 from nose.tools import raises
 
 from remotespark.livyclientlib.livyclientfactory import LivyClientFactory
+from remotespark.livyclientlib.pandaspysparklivyclient import PandasPysparkLivyClient
+from remotespark.livyclientlib.pandasscalalivyclient import PandasScalaLivyClient
 from remotespark.utils.constants import Constants
 from remotespark.utils.utils import get_connection_string
 
@@ -9,36 +11,58 @@ from remotespark.utils.utils import get_connection_string
 def test_build_session_with_defaults():
     factory = LivyClientFactory()
     connection_string = get_connection_string("url", "user", "pass")
-    language = "python"
+    kind = Constants.session_kind_pyspark
+    properties = {"kind": kind}
 
-    session = factory.create_session(language, connection_string)
+    session = factory.create_session(connection_string, properties)
 
-    assert session.language == language
+    assert session.kind == kind
     assert session.id == "-1"
     assert session.started_sql_context is False
+    assert session.properties == properties
 
 
 def test_build_session():
     factory = LivyClientFactory()
     connection_string = get_connection_string("url", "user", "pass")
-    language = "python"
+    kind = Constants.session_kind_pyspark
+    properties = {"kind": kind}
 
-    session = factory.create_session(language, connection_string, "1", True)
+    session = factory.create_session(connection_string, properties, "1", True)
 
-    assert session.language == language
+    assert session.kind == kind
     assert session.id == "1"
     assert session.started_sql_context
+    assert session.properties == properties
 
 
 def test_can_build_all_clients():
-    session = MagicMock()
     factory = LivyClientFactory()
-    for language in Constants.lang_supported:
-        factory.build_client(language, session)
+    for kind in Constants.session_kinds_supported:
+        session = MagicMock()
+        session.kind = kind
+        factory.build_client(session)
 
 
 @raises(ValueError)
 def test_build_unknown_language():
     session = MagicMock()
+    session.kind = "unknown"
+    factory = LivyClientFactory()
+    factory.build_client(session)
+
+
+def test_build_pyspark():
+    session = MagicMock()
+    session.kind = Constants.session_kind_pyspark
+    factory = LivyClientFactory()
+    client = factory.build_client(session)
+    assert isinstance(client, PandasPysparkLivyClient)
+
+
+def test_build_spark():
+    session = MagicMock()
+    session.kind = Constants.session_kind_spark
     factory = LivyClientFactory()
-    factory.build_client("unknown", session)
+    client = factory.build_client(session)
+    assert isinstance(client, PandasScalaLivyClient)
diff --git a/tests/test_livysession.py b/tests/test_livysession.py
index 0090e1b..6ee3d1d 100644
--- a/tests/test_livysession.py
+++ b/tests/test_livysession.py
@@ -7,7 +7,8 @@ from remotespark.livyclientlib.livyclienttimeouterror import LivyClientTimeoutEr
 from remotespark.livyclientlib.livyunexpectedstatuserror import LivyUnexpectedStatusError
 from remotespark.livyclientlib.livysession import LivySession
 import remotespark.utils.configuration as conf
-from remotespark.utils.utils import get_connection_string, get_instance_id
+from remotespark.utils.utils import get_connection_string
+from remotespark.utils.constants import Constants
 
 
 class DummyResponse:
@@ -23,41 +24,22 @@ class DummyResponse:
 
 
 class TestLivySession:
-    pi_result = "Pi is roughly 3.14336"
-
-    session_create_json = '{"id":0,"state":"starting","kind":"spark","log":[]}'
-    ready_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"idle","kind":"spark","log":["16:23:01,15' \
-                          '1 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]' \
-                          '","16:23:01,213 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - As' \
-                          'suming default type [ch.qos.logback.access.PatternLayoutEncoder] for [encoder] propert' \
-                          'y","16:23:01,368 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attachin' \
-                          'g appender named [STDOUT] to null","16:23:01,368 |-INFO in ch.qos.logback.access.joran.act' \
-                          'ion.ConfigurationAction - End of configuration.","16:23:01,371 |-INFO in ch.qos.logback.ac' \
-                          'cess.joran.JoranConfigurator@53799e55 - Registering current configuration as safe fallback' \
-                          ' point","","15/09/04 16:23:01 INFO server.ServerConnector: Started ServerConnector@388859' \
-                          'e4{HTTP/1.1}{0.0.0.0:37394}","15/09/04 16:23:01 INFO server.Server: Started @27514ms","' \
-                          '15/09/04 16:23:01 INFO livy.WebServer: Starting server on 37394","Starting livy-repl on' \
-                          ' http://10.0.0.11:37394"]}]}'
-    error_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"error","kind":"spark","log":[]}]}'
-    busy_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"busy","kind":"spark","log":["16:23:01,151' \
-                         ' |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]",' \
-                         '"16:23:01,213 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assumin' \
-                         'g default type [ch.qos.logback.access.PatternLayoutEncoder] for [encoder] property","16:23' \
-                         ':01,368 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender ' \
-                         'named [STDOUT] to null","16:23:01,368 |-INFO in ch.qos.logback.access.joran.action.Configu' \
-                         'rationAction - End of configuration.","16:23:01,371 |-INFO in ch.qos.logback.access.joran.' \
-                         'JoranConfigurator@53799e55 - Registering current configuration as safe fallback point","",' \
-                         '"15/09/04 16:23:01 INFO server.ServerConnector: Started ServerConnector@388859e4{HTTP/1.1}' \
-                         '{0.0.0.0:37394}","15/09/04 16:23:01 INFO server.Server: Started @27514ms","15/09/04 16:23:' \
-                         '01 INFO livy.WebServer: Starting server on 37394","Starting livy-repl on http://10.0.0.11:' \
-                         '37394"]}]}'
-    post_statement_json = '{"id":0,"state":"running","output":null}'
-    running_statement_json = '{"total_statements":1,"statements":[{"id":0,"state":"running","output":null}]}'
-    ready_statement_json = '{"total_statements":1,"statements":[{"id":0,"state":"available","output":{"status":"ok",' \
-                           '"execution_count":0,"data":{"text/plain":"Pi is roughly 3.14336"}}}]}'
-    
-    get_responses = []
-    post_responses = []
+
+    def __init__(self):
+        self.pi_result = "Pi is roughly 3.14336"
+
+        self.session_create_json = '{"id":0,"state":"starting","kind":"spark","log":[]}'
+        self.ready_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"idle","kind":"spark","log":[""]}]}'
+        self.error_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"error","kind":"spark","log":' \
+                                   '[""]}]}'
+        self.busy_sessions_json = '{"from":0,"total":1,"sessions":[{"id":0,"state":"busy","kind":"spark","log":[""]}]}'
+        self.post_statement_json = '{"id":0,"state":"running","output":null}'
+        self.running_statement_json = '{"total_statements":1,"statements":[{"id":0,"state":"running","output":null}]}'
+        self.ready_statement_json = '{"total_statements":1,"statements":[{"id":0,"state":"available","output":{"statu' \
+                                    's":"ok","execution_count":0,"data":{"text/plain":"Pi is roughly 3.14336"}}}]}'
+
+        self.get_responses = []
+        self.post_responses = []
 
     def _next_response_get(self, *args):
         val = self.get_responses[0]
@@ -69,103 +51,83 @@ class TestLivySession:
         self.post_responses = self.post_responses[1:]    
         return val
 
+    def _create_session(self, kind=Constants.session_kind_spark, session_id="-1", sql_created=False, http_client=None):
+        if http_client is None:
+            http_client = MagicMock()
+
+        return LivySession(http_client, session_id, sql_created, {"kind": kind})
+
     @raises(AssertionError)
     def test_constructor_throws_status_sleep_seconds(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "-1"
-        sql_created = False
         conf.override({
             "status_sleep_seconds": 0,
             "statement_sleep_seconds": 2,
             "create_sql_context_timeout_seconds": 60
         })
-        LivySession(http_client, kind, session_id, sql_created)
+        self._create_session()
         conf.load()
 
     @raises(AssertionError)
     def test_constructor_throws_statement_sleep_seconds(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "-1"
-        sql_created = False
         conf.override({
             "status_sleep_seconds": 3,
             "statement_sleep_seconds": 0,
             "create_sql_context_timeout_seconds": 60
         })
-        LivySession(http_client, kind, session_id, sql_created)
+        self._create_session()
         conf.load({})
 
     @raises(AssertionError)
     def test_constructor_throws_sql_create_timeout_seconds(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "-1"
-        sql_created = False
         conf.override({
             "status_sleep_seconds": 4,
             "statement_sleep_seconds": 2,
             "create_sql_context_timeout_seconds": 0
         })
-        LivySession(http_client, kind, session_id, sql_created)
+        self._create_session()
         conf.load()
 
     @raises(ValueError)
     def test_constructor_throws_invalid_session_sql_combo(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "-1"
-        sql_created = True
         conf.override({
             "status_sleep_seconds": 2,
             "statement_sleep_seconds": 2,
             "create_sql_context_timeout_seconds": 60
         })
-        LivySession(http_client, kind, session_id, sql_created)
+        self._create_session(sql_created=True)
         conf.load()
 
     def test_constructor_starts_with_existing_session(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "1"
-        sql_created = True
         conf.override({
             "status_sleep_seconds": 4,
             "statement_sleep_seconds": 2,
             "create_sql_context_timeout_seconds": 60
         })
-        session = LivySession(http_client, kind, session_id, sql_created)
+        session_id = "1"
+        session = self._create_session(session_id=session_id, sql_created=True)
         conf.load()
 
-        assert session.id == "1"
+        assert session.id == session_id
         assert session.started_sql_context
 
     def test_constructor_starts_with_no_session(self):
-        kind = "scala"
-        http_client = MagicMock()
-        session_id = "-1"
-        sql_created = False
         conf.override({
             "status_sleep_seconds": 4,
             "statement_sleep_seconds": 2,
             "create_sql_context_timeout_seconds": 60
         })
-        session = LivySession(http_client, kind, session_id, sql_created)
+        session = self._create_session()
         conf.load()
 
         assert session.id == "-1"
         assert not session.started_sql_context
 
     def test_is_final_status(self):
-        kind = "scala"
-        http_client = MagicMock()
-
         conf.override({
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session()
         conf.load()
 
         assert not session.is_final_status("idle")
@@ -176,7 +138,6 @@ class TestLivySession:
         assert session.is_final_status("error")
 
     def test_start_scala_starts_session(self):
-        kind = "scala"
         http_client = MagicMock()
         http_client.post.return_value = DummyResponse(201, self.session_create_json)
 
@@ -184,18 +145,18 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        kind = Constants.session_kind_spark
+        session = self._create_session(kind=kind, http_client=http_client)
         session.start()
         conf.load()
 
-        assert_equals(kind, session.language)
+        assert_equals(kind, session.kind)
         assert_equals("starting", session._status)
         assert_equals("0", session.id)
         http_client.post.assert_called_with(
-            "/sessions", [201], {"kind": "spark", "name": "remotesparkmagics_{}".format(get_instance_id())})
+            "/sessions", [201], {"kind": "spark"})
 
     def test_start_python_starts_session(self):
-        kind = "python"
         http_client = MagicMock()
         http_client.post.return_value = DummyResponse(201, self.session_create_json)
 
@@ -203,15 +164,33 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        kind = Constants.session_kind_pyspark
+        session = self._create_session(kind=kind, http_client=http_client)
         session.start()
         conf.load()
 
-        assert_equals(kind, session.language)
+        assert_equals(kind, session.kind)
         assert_equals("starting", session._status)
         assert_equals("0", session.id)
         http_client.post.assert_called_with(
-            "/sessions", [201],{"kind": "pyspark", "name": "remotesparkmagics_{}".format(get_instance_id())})
+            "/sessions", [201], {"kind": "pyspark"})
+
+    def test_start_passes_in_all_properties(self):
+        http_client = MagicMock()
+        http_client.post.return_value = DummyResponse(201, self.session_create_json)
+
+        conf.override({
+            "status_sleep_seconds": 0.01,
+            "statement_sleep_seconds": 0.01
+        })
+        kind = Constants.session_kind_spark
+        properties = {"kind": kind, "extra": 1}
+        session = LivySession(http_client, "-1", False, properties)
+        session.start()
+        conf.load()
+
+        http_client.post.assert_called_with(
+            "/sessions", [201], properties)
 
     def test_status_gets_latest(self):
         http_client = MagicMock()
@@ -221,7 +200,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
         session.start()
 
@@ -242,7 +221,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.override({})
 
         session.start()
@@ -265,7 +244,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.011,
             "statement_sleep_seconds": 6000
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
 
         session.start()
@@ -285,7 +264,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.011,
             "statement_sleep_seconds": 6000
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
 
         session.start()
@@ -299,7 +278,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
         session.start()
 
@@ -315,7 +294,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
 
         session.delete()
@@ -331,14 +310,14 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, "scala", "-1", False)
+        session = self._create_session(http_client=http_client)
         conf.load()
         session._status = "dead"
 
         session.delete()
 
     def test_execute(self):
-        kind = "scala"
+        kind = Constants.session_kind_spark
         http_client = MagicMock()
         self.post_responses = [DummyResponse(201, self.session_create_json),
                                DummyResponse(201, self.post_statement_json)]
@@ -350,7 +329,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
         session.start()
         command = "command"
@@ -364,7 +343,7 @@ class TestLivySession:
         assert_equals(self.pi_result, result[1])
 
     def test_create_sql_hive_context_happens_once(self):
-        kind = "scala"
+        kind = Constants.session_kind_spark
         http_client = MagicMock()
         self.post_responses = [DummyResponse(201, self.session_create_json),
                                DummyResponse(201, self.post_statement_json),
@@ -380,7 +359,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
         session.start()
 
@@ -396,12 +375,11 @@ class TestLivySession:
                                                               "(sc)\nimport sqlContext.implicits._"}) \
                in http_client.post.call_args_list
         assert call("/sessions/0/statements", [201], {"code": "val hiveContext = new org.apache.spark.sql.hive.Hive"
-                                                              "Context(sc)"}) \
-               in http_client.post.call_args_list
+                                                              "Context(sc)"}) in http_client.post.call_args_list
         assert len(http_client.post.call_args_list) == 2
 
     def test_create_sql_context_spark(self):
-        kind = "scala"
+        kind = Constants.session_kind_spark
         http_client = MagicMock()
         self.post_responses = [DummyResponse(201, self.session_create_json),
                                DummyResponse(201, self.post_statement_json),
@@ -417,7 +395,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
         session.start()
 
@@ -427,12 +405,10 @@ class TestLivySession:
                                                               "(sc)\nimport sqlContext.implicits._"}) \
                in http_client.post.call_args_list
         assert call("/sessions/0/statements", [201], {"code": "val hiveContext = new org.apache.spark.sql.hive.Hive"
-                                                              "Context(sc)"}) \
-               in http_client.post.call_args_list
-
+                                                              "Context(sc)"}) in http_client.post.call_args_list
 
     def test_create_sql_hive_context_pyspark(self):
-        kind = "python"
+        kind = Constants.session_kind_pyspark
         http_client = MagicMock()
         self.post_responses = [DummyResponse(201, self.session_create_json),
                                DummyResponse(201, self.post_statement_json),
@@ -448,16 +424,15 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
         session.start()
 
         session.create_sql_context()
 
-        assert call("/sessions/0/statements", [201], {"code": "from pyspark.sql import SQLContext\n"
-                                                              "from pyspark.sql.types import *\n"
-                                                              "sqlContext = SQLContext(sc)"}) \
-               in http_client.post.call_args_list
+        assert call("/sessions/0/statements", [201], {"code": "from pyspark.sql import SQLContext\nfrom pyspark."
+                                                              "sql.types import *\nsqlContext = SQLContext("
+                                                              "sc)"}) in http_client.post.call_args_list
         assert call("/sessions/0/statements", [201], {"code": "from pyspark.sql import HiveContext\n"
                                                               "hiveContext = HiveContext(sc)"}) \
                in http_client.post.call_args_list
@@ -477,7 +452,7 @@ class TestLivySession:
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
         session.start()
 
@@ -490,19 +465,19 @@ class TestLivySession:
         connection_string = get_connection_string(url, username, password)
         http_client = MagicMock()
         http_client.connection_string = connection_string
-        kind = "scala"
+        kind = Constants.session_kind_spark
         conf.override({
             "status_sleep_seconds": 0.01,
             "statement_sleep_seconds": 0.01
         })
-        session = LivySession(http_client, kind, "-1", False)
+        session = self._create_session(kind=kind, http_client=http_client)
         conf.load()
 
         serialized = session.get_state().to_dict()
 
         assert serialized["connectionstring"] == connection_string
         assert serialized["id"] == "-1"
-        assert serialized["language"] == kind
+        assert serialized["kind"] == kind
         assert serialized["sqlcontext"] == False
         assert serialized["version"] == "0.0.0"
         assert len(serialized.keys()) == 5
diff --git a/tests/test_remotesparkmagics.py b/tests/test_remotesparkmagics.py
index e5f6058..8afdc90 100644
--- a/tests/test_remotesparkmagics.py
+++ b/tests/test_remotesparkmagics.py
@@ -3,6 +3,7 @@ from nose.tools import raises, with_setup
 
 from remotespark.remotesparkmagics import RemoteSparkMagics
 from remotespark.livyclientlib.dataframeparseexception import DataFrameParseException
+import remotespark.utils.configuration as conf
 
 
 magic = None
@@ -13,6 +14,8 @@ shell = None
 def _setup():
     global magic, spark_controller, shell
 
+    conf.override({})
+
     shell = MagicMock()
     magic = RemoteSparkMagics(shell=None)
     magic.shell = shell
@@ -38,7 +41,7 @@ def test_info_command_parses():
 
 @with_setup(_setup, _teardown)
 def test_add_sessions_command_parses():
-    # Do not skip
+    # Do not skip and python
     add_sessions_mock = MagicMock()
     spark_controller.add_session = add_sessions_mock
     command = "add"
@@ -49,20 +52,38 @@ def test_add_sessions_command_parses():
 
     magic.spark(line)
 
-    add_sessions_mock.assert_called_once_with(name, language, connection_string, False)
+    add_sessions_mock.assert_called_once_with(name, connection_string, False, {"kind": "pyspark"})
 
-    # Skip
+    # Skip and scala - upper case
     add_sessions_mock = MagicMock()
     spark_controller.add_session = add_sessions_mock
     command = "add"
     name = "name"
-    language = "python"
+    language = "Scala"
     connection_string = "url=http://location:port;username=name;password=word"
     line = " ".join([command, name, language, connection_string, "skip"])
 
     magic.spark(line)
 
-    add_sessions_mock.assert_called_once_with(name, language, connection_string, True)
+    add_sessions_mock.assert_called_once_with(name, connection_string, True, {"kind": "spark"})
+
+
+@with_setup(_setup, _teardown)
+def test_add_sessions_command_extra_properties():
+    magic.spark("config {\"extra\": \"yes\"}")
+    assert magic.properties == {"extra": "yes"}
+
+    add_sessions_mock = MagicMock()
+    spark_controller.add_session = add_sessions_mock
+    command = "add"
+    name = "name"
+    language = "scala"
+    connection_string = "url=http://location:port;username=name;password=word"
+    line = " ".join([command, name, language, connection_string])
+
+    magic.spark(line)
+
+    add_sessions_mock.assert_called_once_with(name, connection_string, False, {"kind": "spark", "extra": "yes"})
 
 
 @with_setup(_setup, _teardown)
diff --git a/tests/test_sparkcontroller.py b/tests/test_sparkcontroller.py
index fcd1ff4..8d886fc 100644
--- a/tests/test_sparkcontroller.py
+++ b/tests/test_sparkcontroller.py
@@ -25,17 +25,17 @@ def _teardown():
 @with_setup(_setup, _teardown)
 def test_add_session():
     name = "name"
-    language = "python"
+    properties = {"kind": "spark"}
     connection_string = "url=http://location:port;username=name;password=word"
     client = "client"
     session = MagicMock()
     client_factory.create_session = MagicMock(return_value=session)
     client_factory.build_client = MagicMock(return_value=client)
 
-    controller.add_session(name, language, connection_string, False)
+    controller.add_session(name, connection_string, False, properties)
 
-    client_factory.create_session.assert_called_once_with(language, connection_string, "-1", False)
-    client_factory.build_client.assert_called_once_with(language, session)
+    client_factory.create_session.assert_called_once_with(connection_string, properties, "-1", False)
+    client_factory.build_client.assert_called_once_with(session)
     client_manager.add_client.assert_called_once_with(name, client)
     session.start.assert_called_once_with()
 

pip install -e .
parse_log_pytest
requirements.txt
pytest
mkdir ~/.sparkmagic cp remotespark/default_config.json ~/.sparkmagic/config.json
3.9
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
anyio==4.9.0
argon2-cffi==23.1.0
argon2-cffi-bindings==21.2.0
arrow==1.3.0
async-lru==2.0.5
attrs==25.3.0
babel==2.17.0
beautifulsoup4==4.13.3
bleach==6.2.0
certifi==2025.1.31
cffi==1.17.1
charset-normalizer==3.4.1
comm==0.2.2
decorator==5.2.1
defusedxml==0.7.1
exceptiongroup==1.2.2
fastjsonschema==2.21.1
fqdn==1.5.1
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
idna==3.10
importlib_metadata==8.6.1
iniconfig==2.1.0
ipykernel==4.1.1
ipython==4.0.0
ipython-genutils==0.2.0
ipywidgets==7.8.5
isoduration==20.11.0
Jinja2==3.1.6
json5==0.10.0
jsonpointer==3.0.0
jsonschema==4.23.0
jsonschema-specifications==2024.10.1
jupyter-events==0.12.0
jupyter-lsp==2.2.5
jupyter_client==8.6.3
jupyter_core==5.7.2
jupyter_server==2.15.0
jupyter_server_terminals==0.5.3
jupyterlab==4.1.5
jupyterlab_pygments==0.3.0
jupyterlab_server==2.27.3
jupyterlab_widgets==1.1.11
MarkupSafe==3.0.2
mistune==3.1.3
mock==5.2.0
narwhals==1.32.0
nbclient==0.10.2
nbconvert==7.16.6
nbformat==5.10.4
nose==1.3.7
notebook==7.1.3
notebook_shim==0.2.4
numpy==2.0.2
overrides==7.7.0
packaging==24.2
pandas==2.2.3
pandocfilters==1.5.1
pexpect==4.9.0
pickleshare==0.7.5
platformdirs==4.3.7
plotly==6.0.1
pluggy==1.5.0
prometheus_client==0.21.1
ptyprocess==0.7.0
pycparser==2.22
Pygments==2.19.1
pytest==8.3.5
python-dateutil==2.9.0.post0
python-json-logger==3.3.0
pytz==2025.2
PyYAML==6.0.2
pyzmq==26.3.0
referencing==0.36.2
-e git+https://github.com/jupyter-incubator/sparkmagic.git@3c0230b30b1f63780fbf2d4ba9d83e0a83f51eea#egg=remotespark
requests==2.32.3
rfc3339-validator==0.1.4
rfc3986-validator==0.1.1
rpds-py==0.24.0
Send2Trash==1.8.3
simplegeneric==0.8.1
six==1.17.0
sniffio==1.3.1
soupsieve==2.6
terminado==0.18.1
tinycss2==1.4.0
tomli==2.2.1
tornado==6.4.2
traitlets==5.14.3
types-python-dateutil==2.9.0.20241206
typing_extensions==4.13.0
tzdata==2025.2
uri-template==1.3.0
urllib3==2.3.0
webcolors==24.11.1
webencodings==0.5.1
websocket-client==1.8.0
widgetsnbextension==3.6.10
zipp==3.21.0

name: sparkmagic
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - anyio==4.9.0
      - argon2-cffi==23.1.0
      - argon2-cffi-bindings==21.2.0
      - arrow==1.3.0
      - async-lru==2.0.5
      - attrs==25.3.0
      - babel==2.17.0
      - beautifulsoup4==4.13.3
      - bleach==6.2.0
      - certifi==2025.1.31
      - cffi==1.17.1
      - charset-normalizer==3.4.1
      - comm==0.2.2
      - decorator==5.2.1
      - defusedxml==0.7.1
      - exceptiongroup==1.2.2
      - fastjsonschema==2.21.1
      - fqdn==1.5.1
      - h11==0.14.0
      - httpcore==1.0.7
      - httpx==0.28.1
      - idna==3.10
      - importlib-metadata==8.6.1
      - iniconfig==2.1.0
      - ipykernel==4.1.1
      - ipython==4.0.0
      - ipython-genutils==0.2.0
      - ipywidgets==7.8.5
      - isoduration==20.11.0
      - jinja2==3.1.6
      - json5==0.10.0
      - jsonpointer==3.0.0
      - jsonschema==4.23.0
      - jsonschema-specifications==2024.10.1
      - jupyter-client==8.6.3
      - jupyter-core==5.7.2
      - jupyter-events==0.12.0
      - jupyter-lsp==2.2.5
      - jupyter-server==2.15.0
      - jupyter-server-terminals==0.5.3
      - jupyterlab==4.1.5
      - jupyterlab-pygments==0.3.0
      - jupyterlab-server==2.27.3
      - jupyterlab-widgets==1.1.11
      - markupsafe==3.0.2
      - mistune==3.1.3
      - mock==5.2.0
      - narwhals==1.32.0
      - nbclient==0.10.2
      - nbconvert==7.16.6
      - nbformat==5.10.4
      - nose==1.3.7
      - notebook==7.1.3
      - notebook-shim==0.2.4
      - numpy==2.0.2
      - overrides==7.7.0
      - packaging==24.2
      - pandas==2.2.3
      - pandocfilters==1.5.1
      - pexpect==4.9.0
      - pickleshare==0.7.5
      - platformdirs==4.3.7
      - plotly==6.0.1
      - pluggy==1.5.0
      - prometheus-client==0.21.1
      - ptyprocess==0.7.0
      - pycparser==2.22
      - pygments==2.19.1
      - pytest==8.3.5
      - python-dateutil==2.9.0.post0
      - python-json-logger==3.3.0
      - pytz==2025.2
      - pyyaml==6.0.2
      - pyzmq==26.3.0
      - referencing==0.36.2
      - requests==2.32.3
      - rfc3339-validator==0.1.4
      - rfc3986-validator==0.1.1
      - rpds-py==0.24.0
      - send2trash==1.8.3
      - simplegeneric==0.8.1
      - six==1.17.0
      - sniffio==1.3.1
      - soupsieve==2.6
      - terminado==0.18.1
      - tinycss2==1.4.0
      - tomli==2.2.1
      - tornado==6.4.2
      - traitlets==5.14.3
      - types-python-dateutil==2.9.0.20241206
      - typing-extensions==4.13.0
      - tzdata==2025.2
      - uri-template==1.3.0
      - urllib3==2.3.0
      - webcolors==24.11.1
      - webencodings==0.5.1
      - websocket-client==1.8.0
      - widgetsnbextension==3.6.10
      - zipp==3.21.0
prefix: /opt/conda/envs/sparkmagic

tests/test_clientmanagerstateserializer.py::test_deserialize_not_emtpy tests/test_livyclient.py::test_serialize tests/test_livyclient.py::test_kind tests/test_livyclientfactory.py::test_build_session_with_defaults tests/test_livyclientfactory.py::test_build_session tests/test_livyclientfactory.py::test_can_build_all_clients tests/test_livyclientfactory.py::test_build_unknown_language tests/test_livyclientfactory.py::test_build_pyspark tests/test_livyclientfactory.py::test_build_spark
tests/test_clientmanagerstateserializer.py::test_deserialize_not_emtpy_but_dead tests/test_clientmanagerstateserializer.py::test_deserialize_not_emtpy_but_error tests/test_clientmanagerstateserializer.py::test_deserialize_empty tests/test_remotesparkmagics.py::test_info_command_parses tests/test_remotesparkmagics.py::test_add_sessions_command_parses tests/test_remotesparkmagics.py::test_add_sessions_command_extra_properties tests/test_remotesparkmagics.py::test_delete_sessions_command_parses tests/test_remotesparkmagics.py::test_cleanup_command_parses tests/test_remotesparkmagics.py::test_bad_command_throws_exception tests/test_remotesparkmagics.py::test_run_cell_command_parses tests/test_remotesparkmagics.py::test_run_cell_command_writes_to_err tests/test_remotesparkmagics.py::test_run_sql_command_parses tests/test_remotesparkmagics.py::test_run_hive_command_parses tests/test_remotesparkmagics.py::test_run_sql_command_returns_none_when_exception tests/test_remotesparkmagics.py::test_run_hive_command_returns_none_when_exception tests/test_remotesparkmagics.py::test_run_sql_command_stores_variable_in_user_ns tests/test_sparkcontroller.py::test_add_session tests/test_sparkcontroller.py::test_add_session_skip tests/test_sparkcontroller.py::test_delete_session tests/test_sparkcontroller.py::test_cleanup tests/test_sparkcontroller.py::test_run_cell tests/test_sparkcontroller.py::test_get_client_keys
tests/test_clientmanagerstateserializer.py::test_serializer_throws_none_path tests/test_clientmanagerstateserializer.py::test_serializer_throws_none_factory tests/test_clientmanagerstateserializer.py::test_serialize_not_empty tests/test_livyclient.py::test_create_sql_context_automatically tests/test_livyclient.py::test_execute_code tests/test_livyclient.py::test_execute_sql tests/test_livyclient.py::test_execute_hive tests/test_livyclient.py::test_close_session tests/test_livyclient.py::test_session_id
remotespark/livyclientlib/livysessionstate.py remotespark/remotesparkmagics.py remotespark/utils/configuration.py remotespark/livyclientlib/livyclient.py remotespark/livyclientlib/clientmanagerstateserializer.py remotespark/default_config.json remotespark/livyclientlib/livysession.py remotespark/livyclientlib/livyclientfactory.py remotespark/livyclientlib/sparkcontroller.py remotespark/utils/constants.py
remotespark/livyclientlib/livysessionstate.py remotespark/remotesparkmagics.py remotespark/utils/configuration.py remotespark/livyclientlib/livyclient.py remotespark/livyclientlib/clientmanagerstateserializer.py remotespark/default_config.json remotespark/livyclientlib/livysession.py remotespark/livyclientlib/livyclientfactory.py remotespark/livyclientlib/sparkcontroller.py remotespark/utils/constants.py