Optimization+Modeling+and+Relational+Data+pub.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Optimization Modeling and Relational Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook shows the relationship between mathematical models used in optimization and data models used to store and retrieve the data that populates a model instance. It illustrates, by means of an example, how the data structures of the OPL modeling language used in optimization can be constructed using SQL, focusing specifically on how to use Spark dataframes for this purpose.\n",
    "<p>\n",
    "In this notebook, you will learn how to set up the optimization problem using IBM's OPL modeling language and how to solve it using IBM's Decision Optimization on Cloud service. The notebook also shows you how access data from a source in IBM's DSX Community and how to use Apache Spark to manage the data input to and output from the optimization service.\n",
    "<p>\n",
    ">This notebook is part of [IBM Decision Optimization on Cloud service with the Python Client ](https://developer.ibm.com/docloud/documentation/docloud/python-api/).\n",
    "\n",
    ">You will need a valid subscription to Decision Optimization on Cloud ([here](https://developer.ibm.com/docloud)). \n",
    "\n",
    "Some familiarity with Python is recommended. This notebook runs on Python 2 with Spark 2.0."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Table of Contents"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [1 Introduction](#1-Introduction)\n",
    "\t- [1.1 The Tableau Form of an Optimization Model](#1.1-The-Tableau-Form-of-an-Optimization-Model)\n",
    "- [2 An Example – Warehouse Location](#2-An-Example-–-Warehouse-Location)\n",
    "\t- [2.1 The Business Context](#2.1-The-Business-Context)\n",
    "\t- [2.2 The Application Data Model](#2.2-The-Application-Data-Model)\n",
    "\t- [2.3 The OPLCollector Class](#2.3-The-OPLCollector-Class)\n",
    "\t- [2.4 The Spark Data Model for the Warehouse Location Application](#2.4-The-Spark-Data-Model-for-the-Warehouse-Location-Application)\n",
    "\t- [2.5 The Data](#2.5-The-Data)\n",
    "\t- [2.6 Optimization Model](#2.6-Optimization-Model)\n",
    "\t- [2.7 Solving the Warehousing Model with IBM Decision Optimization on Cloud](#2.7-Solving-the-Warehousing-Model-with-IBM-Decision-Optimization-on-Cloud)\n",
    "\t\t- [2.7.1 Get Your Credentials for IBM Decision Optimization on Cloud](#2.7.1-Get-Your-Credentials-for-IBM-Decision-Optimization-on-Cloud)\n",
    "\t\t- [2.7.2 The Optimizer Class](#2.7.2-The-Optimizer-Class)\n",
    "\t\t- [2.7.3 Setting Up and Submitting the Solve Job](#2.7.3-Setting-Up-and-Submitting-the-Solve-Job)\n",
    "\t\t- [2.7.4 Retrieving the Optimal Solution](#2.7.4-Retrieving-the-Optimal-Solution)\n",
    "- [3 Transforming an Optimization Problem to Tableau Form](#3-Transforming-an-Optimization-Problem-to-Tableau-Form)\n",
    "\t- [3.1 Using Relational Database Operations to Reshape the Instance Data](#3.1-Using-Relational-Database-Operations-to-Reshape-the-Instance-Data)\n",
    "\t\t- [3.1.1 The Tableau Data Model](#3.1.1-The-Tableau-Data-Model)\n",
    "\t\t- [3.1.2 The Transformation Data Model](#3.1.2-The-Transformation-Data-Model)\n",
    "\t- [3.2 Encoding the Decision Variables](#3.2-Encoding-the-Decision-Variables)\n",
    "\t- [3.3 Encoding the Constraints and Decision Expressions](#3.3-Encoding-the-Constraints-and-Decision-Expressions)\n",
    "\t- [3.4 Reshaping the Coefficient Data into the Tableau](#3.4-Reshaping-the-Coefficient-Data-into-the-Tableau)\n",
    "\t- [3.5 Creating the Tableau Input Data](#3.5-Creating-the-Tableau-Input-Data)\n",
    "\t- [3.6 Solving the Tableau Model](#3.6-Solving-the-Tableau-Model)\n",
    "\t- [3.7 Recovering the Warehousing Solution](#3.7-Recovering-the-Warehousing-Solution)\n",
    "- [4 Conclusion: Equivalence of Tuple Slicing and SQL](#4-Conclusion:-Equivalence-of-Tuple-Slicing-and-SQL)\n",
    "- [Author](#Author)\n",
    "- [References](#References)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1 Introduction"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The purpose of this notebook is to explicate the relationship between mathematical models used in optimization and data models used to store and retrieve the data that populates a model instance. \n",
    "Business solutions based on predictive analytics typically have a three-layer architecture:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "figure1= \"\"\"\n",
    "                                         Extract,\n",
    "                                         Validate,                 Visualize,\n",
    "                                         Transform                 Interact\n",
    "                        +---------------+         +---------------+        +---------------+\n",
    "                        |               +-------> +               +------> +               |\n",
    "                        |  Enterprise   |         |  Application  |        |   Business    |\n",
    "                        |  Data         |         |  Layer        |        |   Interface   |\n",
    "                        |               + <-------+               + <------+               |\n",
    "                        +---------------+         +----+-----+----+        +---------------+\n",
    "                                                       |     ^\n",
    "                                                       |     |\n",
    "                                                       |     |  Application Data Model\n",
    "                                                       |     |\n",
    "                                                       v     |\n",
    "                                                  +----+-----+----+\n",
    "                                                  |               |\n",
    "                                                  |   Modeling    |\n",
    "                                                  |   Layer       |\n",
    "                                                  |               |\n",
    "                                                  +----+-----+----+\n",
    "                                                       |     ^\n",
    "                                                       |     |\n",
    "                                                       |     |  Tableau Data Model\n",
    "                                                       |     |\n",
    "                                                       v     |\n",
    "                                                  +----+-----+----+\n",
    "                                                  |               |\n",
    "                                                  |    Solving    |\n",
    "                                                  |    Layer      |\n",
    "                                                  |               |\n",
    "                                                  +---------------+\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The application layer provides the business logic of the solution, encompassing such things as data management (exchange with enterprise sources, validation, and transformation), business process workflow, interaction with business users, and so forth. The modeling layer represents the underlying business problem mathematically. The solving layer executes the optimization algorithms on the mathematical model. This paper focuses on the modeling layer and its relationship with the solving layer."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Given the large sizes of data sets used in optimization and the large-scale capabilities of modern mathematical programming solvers, a primary objective of the modeling layer is to move data efficiently and quickly between external databases and the internal data structures of the solver. For the purposes of this discussion, it is assumed that that the data exists in the form of relational tables and that the solver works with a (sparse) matrix representation of the optimization problem."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The general form of the optimization problem is linear:\n",
    "\n",
    "\\begin{align}\n",
    "&\\min z=\\sum_{j\\in J} c_{j}x_{j}\\\\\n",
    "&subject \\; to:\\\\\n",
    "&\\sum_{j\\in J}a_{ij}x_{j}\\leq b_{i} \\quad \\forall{i\\in I}\\\\\n",
    "&l_{j}\\leq x_{j}\\leq u_{j} \\quad \\forall{j\\in J}\\\\\n",
    "\\end{align}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The formulation could add boolean and integer variables, as will be discussed below, and it could include special structure constraints, such as generalized upper bounds, but these are not particularly germane to the discussion. Adding non-linear constraints, however, raises the issue of how to represent the non-linear functions, which is beyond the scope of this discussion."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The fundamental data structure underlying this optimization problem is the *simplex tableau*, which consists of the coefficient matrix $a_{ij}$, the cost and right-hand side (*rhs*) vectors $c_{j}$ and $b_{i}$, and the lower and upper bound vectors $l_{j}$ and $u_{j}$. In most real world applications of decision optimization, the tableau is very sparse; that is, very few of its coefficients are non-zero, often fewer than 1%. The algorithms for solving the optimization are especially structured to take advantage of this sparsity. Thus, only entries for which $a_{ij}≠0$ are explicitly represented. Ordinarily, the indices $i$ and $j$ are taken to be integers and the index sets $I$ and $J$ are taken to be sets of integers. This paper will call this formulation the *tableau* representation, in homage to George Dantzig, the inventor of linear programming and the Simplex algorithm for solving them. (See [References](#References) below. Dantzig himself credits the economist Quesnay with coining the term *tableau* in his 1759 book.) In this case, the transformation of the model into the internal matrix representation for the solver is straightforward; each index $j$ corresponds to a column and each $i$ corresponds to a row. However, integer indexing is often too restrictive in real-world optimization problems. Most large optimization models have a great deal of structure, with multiple classes of decision variables and constraints, each of which has its own indexing scheme. The modeling layer then needs to handle the mapping between the index sets and the rows and columns of the tableau representation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Authors of optimization models have long recognized the value of using a more general formulation of an optimization problem. In the more general representation, the indices *i* and *j* are taken to be tuples. A *tuple* is multicomponent data structure where each component, or *field*, is a simple data type, such as an integer or a character string. A subset of the fields of the tuple that uniquely identifies it is called the *key*; the data in the key fields must be discrete. The data in the non-key fields, on the other hand, may be continuous, e.g. a floating point number. (Note that, while there are similarities between the tuples discussed in this paper and the similarly named Python objects, it is important that they remain distinct.) Here is an example of a tuple that might arise in a network flow optimization:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<code>tuple Route {\n",
    " key string location;\n",
    " key string store;\n",
    " float shippingCost;\t// $/pallet\n",
    "}</code>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The preceding example is written in the well-known Optimization Programming Language, or *OPL*, one of a number of specialized languages for  expressing optimization problems (see __[IBM Knowledge Center](http://www.ibm.com/support/knowledgecenter/SSSA5P_12.6.1/ilog.odms.ide.help/OPL_Studio/maps/groupings/opl_Language.html)__). Henceforth, further examples will be displayed in OPL, rather than in mathematical notation. Using OPL has the advantage of producing executable code; how to call OPL from Python will be discussed in section 2.7 below. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1 The Tableau Form of an Optimization Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In OPL, the general optimization problem has the following form:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "tableau_data_model = '''\n",
    " //Defines a column for a boolean variable\n",
    " tuple BooleanColumn {\n",
    "    key string variable;    //name = variable+index\n",
    "    int lower;              //lower bound (always 0)\n",
    "    int upper;              //upper bound (always 1)\n",
    "    int value;              //optimal value (output only)\n",
    " }\n",
    " \n",
    " //Defines a column for an integer variable\n",
    " tuple IntegerColumn {\n",
    "    key string variable;    //name = variable+index\n",
    "    int lower;              //lower bound\n",
    "    int upper;              //upper bound\n",
    "    int value;              //optimal value (output only)\n",
    " }\n",
    " \n",
    " //Defines a column for a continuous variable\n",
    " tuple FloatColumn {\n",
    "    key string variable;    //name = variable+index\n",
    "    float lower;            //lower bound\n",
    "    float upper;            //upper bound\n",
    "    float value;            //optimal value (output only)\n",
    " }\n",
    " \n",
    " //Defines a row for a constraint or a decision expression\n",
    " tuple Row {\n",
    "    key string cnstraint;   //name = constraint+index  \n",
    "    string sense;           //GE(>=), EQ(==), LE(<=) or dexpr (must be dexpr for a decision expression)\n",
    "    float rhs;              //right-hand side term (must be zero for a decision expression)\n",
    " }\n",
    "  \n",
    " //Defines a coefficient at a specific row and column\n",
    " tuple Entry {\n",
    "    key string cnstraint;   //name = constraint+index   (can also be used for a decision expression) \n",
    "    key string variable;    //name = variable+index  \n",
    "    float coefficient;      //coefficent\n",
    " }\n",
    " \n",
    " //Defines a decision expression value\n",
    " tuple Objective {\n",
    "    key string name;        //must correspond to the name of one of the decision expressions\n",
    "    string sense;           //minimize or maximize\n",
    "    float value;            //optimal decision expression value (output only)\n",
    " }\n",
    " '''\n",
    "tableau_inputs = '''\n",
    " {BooleanColumn} \tbooleanColumns= ...;\n",
    " {IntegerColumn} \tintegerColumns=  ...; \n",
    " {FloatColumn} \t\tfloatColumns=  ...; \n",
    " {Row} \t\t\t\trows= ...; \n",
    " {Entry} \t\t\tentries= ...;  \n",
    " {Objective}\t\tobjectives= ...;\t//a singleton tuple set designating the decison express to use as the objective function\n",
    " float\t\t\t\tobjectiveSense= (first(objectives).sense==\"maximize\" ? 1.0 : -1.0);\n",
    " {Row} \t\t\t\trows_dexpr= {i| i in rows: i.sense==\"dexpr\"};\n",
    " '''\n",
    " \n",
    "tableau_optimization_problem='''\n",
    " dvar boolean\tx[booleanColumns];\n",
    " dvar int\t\ty[j in integerColumns]\tin j.lower..j.upper;\n",
    " dvar float\t\tz[j in floatColumns]\tin j.lower..j.upper;\n",
    " \n",
    " dexpr float v[i in rows_dexpr]= \t\t\t \t\n",
    " \t\t\t\t  sum(j in booleanColumns,   t in entries: t.cnstraint==i.cnstraint && t.variable==j.variable) t.coefficient*x[j] \n",
    " \t\t\t\t+ sum(j in integerColumns,   t in entries: t.cnstraint==i.cnstraint && t.variable==j.variable) t.coefficient*y[j] \n",
    "\t\t\t \t+ sum(j in floatColumns,     t in entries: t.cnstraint==i.cnstraint && t.variable==j.variable) t.coefficient*z[j];\n",
    "\n",
    " dexpr float obj= sum(i in rows_dexpr: i.cnstraint==first(objectives).name)v[i]; //selects the decision expression designated as the objective function\n",
    "\n",
    " constraint ct[rows];\n",
    "\n",
    " maximize obj*objectiveSense;\n",
    " subject to {\n",
    "\n",
    " forall(i in rows)\n",
    "   ct[i]:\tif(i.sense==\"GE\")   \n",
    "\t   \t\t\tsum(j in booleanColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*x[j] +\n",
    "\t   \t\t\tsum(j in integerColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*y[j] +\n",
    "\t   \t\t\tsum(j in floatColumns,\t\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*z[j]\n",
    "\t   \t\t\t>= i.rhs;\n",
    "   \t\t\telse if(i.sense==\"EQ\")\n",
    "\t   \t\t\tsum(j in booleanColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*x[j] +\n",
    "\t   \t\t\tsum(j in integerColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*y[j] +\n",
    "\t   \t\t\tsum(j in floatColumns, \t \tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*z[j]\n",
    "\t   \t\t\t== i.rhs;\n",
    "   \t\t\telse if(i.sense==\"LE\")\n",
    "\t   \t\t\tsum(j in booleanColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*x[j] +\n",
    "\t   \t\t\tsum(j in integerColumns,\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*y[j] +\n",
    "\t   \t\t\tsum(j in floatColumns,\t\tt in entries: t.cnstraint==i.cnstraint && t.variable==j.variable)  t.coefficient*z[j]\n",
    "\t   \t\t\t<= i.rhs;\n",
    "\t\t\t\t\n",
    " }\n",
    " '''\n",
    "tableau_outputs='''\n",
    " {BooleanColumn}\tbooleanDecisions=\t\t{<j.variable, j.lower, j.upper, x[j]> | j in booleanColumns};\n",
    " {IntegerColumn}\tintegerDecisions=\t\t{<j.variable, j.lower, j.upper, y[j]> | j in integerColumns};\n",
    " {FloatColumn}\t\tfloatDecisions=\t\t\t{<j.variable, j.lower, j.upper, z[j]> | j in floatColumns};\n",
    " {Objective}\t\toptimalObjectives=\t\t{<i.cnstraint, \"\", v[i]> | i in rows_dexpr};\n",
    " '''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This formulation is more general than the mathematics shown in section 1, in the following ways:\n",
    "- allows decision variables of types <code>boolean</code> (\"yes\" or \"no\"), <code>int</code> (integer), and <code>float</code> (double precision)\n",
    "- allows constraints with senses less than or equal to, equal to, and greater than or equal to\n",
    "- allows multiple decision expressions, one of which serves as the objective function\n",
    "- allows either minimization or maximization of the objective function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This formulation takes a modest step towards the tuple representation, using <code>BooleanColumn</code> and <code>FloatColumn</code> to associate an index string with each variable in the general optimization problem, and similarly for the constraints. More importantly, the tuple representation of the tableau model is sparse, requiring an <code>Entry</code> only where the <code>coefficient</code> field is non-zero. OPL permits representing a matrix as <code>float a[Row][Column]</code>, but this representation would entail reading an entry for every row and column, a much larger dataset when, typically, fewer than 10% of the entries are non-zero."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that the solution takes the form of arrays rather than tuple sets:\n",
    "<code>\n",
    " dvar boolean   x[booleanColumns];\n",
    " dvar int       y[j in integerColumns]  in j.lower..j.upper;\n",
    " dvar float     z[j in floatColumns]    in j.lower..j.upper;\n",
    "</code>\n",
    "\n",
    "Thus it is necessary to reshape the solution into tuple sets as shown in <code>tableau_outputs</code>."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is important to realize that the tableau model represents *any* linear mathematical optimization model (possibly with extensions to incorporate special structure constraints such as generalized upper bounds and others). It actually embodies the low-level interface to the solver and, thus, is properly part of the solving layer rather than the modeling layer. The remainder of this paper focuses on how the modeling layer translates a specific model instance into the generic tableau representation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2 An Example &ndash; Warehouse Location"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook uses an example to make the concepts discussed more concrete. This example is more fully explored in the notebook *Locating Warehouses to Minimize Costs Case 1*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 The Business Context"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A consumer packaged goods supplier needs to decide where to locate its warehouses to serve a set of retail stores at different locations. At the same time, it also needs to determine how much capacity each warehouse should have. The cost of opening a warehouse has a fixed component, related to the acquisition of land and designing the facility, and a variable component proportional to the capacity of the warehouse. The cost to ship the goods from a warehouse to a store depends on the distance between them. The objective is to minimize the cost of opening the warehouses and shipping the goods. Such an optimization application would typically be used as part of an annual planning process in which the company’s management would decide on sales targets and the capital investments needed to support them."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 The Application Data Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The application data model is the schema of the data input to and output from the optimization. It is typically realized in several forms: as the table schema of a relational database system, as the tuple structure of the optimization model, or as a set of classes in a programming language. Here is the OPL representation of the application data model for the Warehousing optimization model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "warehousing_data_dotmod = '''\n",
    " //Input data\n",
    " \n",
    " tuple Warehouse {\n",
    " \tkey string location;\n",
    " \tfloat fixedCost;\t// $/yr\n",
    " \tfloat capacityCost;\t// $/pallet/yr\n",
    " }\n",
    " \n",
    " tuple Store {\n",
    " \tkey string storeId; \n",
    " }\n",
    " \n",
    " tuple Route {\n",
    " \tkey string location;\n",
    " \tkey string store;\n",
    " \tfloat shippingCost;\t// $/pallet\n",
    " }\n",
    " \n",
    " //Note: the mapCoordinates table is not used in the optimization and so is not sent to the optimizer\n",
    "\n",
    " tuple Demand {\n",
    " \tkey string store;\n",
    " \tkey string scenarioId;\n",
    " \tfloat amount;\t\t// pallets/period\n",
    " }\n",
    " \n",
    "  tuple Scenario {\n",
    " \tkey string id;\n",
    " \tfloat totalDemand;\n",
    " \tfloat periods; \t//the number of periods per year during which this scenario prevails; periods = scenario probability * total periods/year\n",
    " }\n",
    " \n",
    " //Output data\n",
    " \n",
    "  tuple Objective {\n",
    "\tkey string problem;\n",
    " \tkey string dExpr;\n",
    " \tkey string scenarioId;\n",
    "\tkey int iteration;\n",
    "\tfloat value;   \n",
    " }\n",
    " \n",
    " tuple Shipment {\n",
    "  \tkey string location;\n",
    " \tkey string store;\n",
    " \tkey string scenarioId;\n",
    " \tkey int iteration;\n",
    " \tfloat amount; \n",
    " }\n",
    " \n",
    " tuple OpenWarehouse {\n",
    " \tkey string location;\n",
    " \tkey string scenarioId;\n",
    " \tkey int iteration;\n",
    " \tint open;\n",
    " \tfloat capacity;\t\t// pallets\n",
    " }\n",
    " '''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 The OPLCollector Class"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to work with the application data, we use a couple of objects (one for input, the other for output) of a class called the <code>OPLCollector</code>. These objects hold the data as Spark datasets. \n",
    "<p>\n",
    "The <code>OPLCollector</code> class itself does not require customization for each application. Instead, it is configured by specifying the schemas of the tables it contains, using a builder method, as will be shown below. The schemas themselves are instances of the Spark <code>StructType</code> class. The design of the <code>OPLCollector</code> class minimizes the amount of custom coding required to build an optimization-based application. (Note that, despite its name, <code>OPLCollector</code> has no dependence on the OPL modeling language and can be used with the DOCplex Python modeling language as well.) Here is the Python code for <code>OPLCollector</code> and some related functions (go to edit mode to see the contents of this hidden cell)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# @hidden_cell\n",
    "\n",
    "'''\n",
    "Created on Feb 8, 2017\n",
    "@author: bloomj\n",
    "'''\n",
    "import sys\n",
    "import os\n",
    "import json\n",
    "import requests\n",
    "\n",
    "try:\n",
    "    from pyspark import SparkConf, SparkContext\n",
    "    from pyspark.sql import SparkSession, Row, functions\n",
    "    from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType\n",
    "except ImportError as e:\n",
    "    print (\"Error importing Spark Modules\", e)\n",
    "    sys.exit(1)\n",
    "\n",
    "SPARK_CONTEXT = sc # sc is predefined\n",
    "SQL_CONTEXT = sqlContext  # sqlContext is predefined\n",
    "SPARK_SESSION = SparkSession.builder.config(\"spark.sql.crossJoin.enabled\", \"true\").getOrCreate()\n",
    "\n",
    "\n",
    "class OPLCollector(object):\n",
    "    '''\n",
    "    Represents an OPL data model in Spark.\n",
    "    Note: Use of this class does not depend on OPL, and in particular, it can be used with the DOcplex Python API.\n",
    "    An application data model (ADM) consists of a set of tables (OPL Tuplesets), each with its own schema.\n",
    "    An ADM is represented by a dictionary in which the keys are the table names and the values are the table schemas.\n",
    "    A builder is provided to create the ADM.\n",
    "\n",
    "    The OPLCollector holds the actual data in Spark Datasets. There are several ways to populate\n",
    "    the data.\n",
    "    - Spark SQL operations can transform tables into other tables.\n",
    "    - A builder is provided when the data is generated programmatically.\n",
    "    - JSON deserialization and serialization are provided when data is exchanged with external applications or stores.\n",
    "\n",
    "    The design of the OPLCollector class aims to reduce the amount of data that must be\n",
    "    manipulated outside of Spark. Where possible, data is streamed among applications without\n",
    "    creating auxiliary in-memory structures or files.\n",
    "\n",
    "    The design of OPLCollector also aims to minimize the amount\n",
    "    of custom coding required to build an application. Collectors are configured\n",
    "    by specifying their schemas through builders rather than by extending with subclasses.\n",
    "    '''\n",
    "\n",
    "    def __init__(self, collectorName, applicationDataModel={}, sparkData={}):\n",
    "        '''\n",
    "        Creates a new OPLCollector instance.\n",
    "\n",
    "        :param collectorName: the name of this collector.\n",
    "        :type collectorName: String\n",
    "        :param applicationDataModel: holds the table schemas for this collector. Each schema is a Spark StructType.\n",
    "        Note that each collector has one and only one application data model.\n",
    "        :type applicationDataModel: dict<String, StructType>\n",
    "        :param sparkData: holds the actual data tables of this collector as a set of Spark datasets.\n",
    "        :type sparkData: dict<String, Dataframe>\n",
    "        '''\n",
    "        self.name = collectorName\n",
    "        self.applicationDataModel = applicationDataModel\n",
    "        self.sparkData = sparkData\n",
    "        self.size = {name: None for name in applicationDataModel.keys()}\n",
    "        self.jsonDestination = None\n",
    "        self.jsonSource = None\n",
    "\n",
    "    def copy(self, name):\n",
    "        \"\"\"\n",
    "        Creates a new OPLCollector instance with copies of the application data model and Spark datasets of this collector.\n",
    "        The ADM copy is immutable. The Spark datasets themselves are immutable, but the copy supports the addTable, addData, and replaceTable methods.\n",
    "        Does not copy the JSONSource or JSONDestination fields.\n",
    "\n",
    "        :param name of the new collector\n",
    "        :param tableNames tables to be copied (all tables in this collector, if absent)\n",
    "        :return a new OPLCollector instance\n",
    "        \"\"\"\n",
    "        result = OPLCollector(name, self.applicationDataModel, self.sparkData.copy())\n",
    "        result.size = self.size.copy()\n",
    "        return result\n",
    "\n",
    "    def copy(self, name, *tables):\n",
    "        result = OPLCollector(name)\n",
    "        admBuilder = ADMBuilder(result);\n",
    "        for table in tables:\n",
    "            admBuilder.addSchema(table, self.getSchema(table))\n",
    "        admBuilder.build()\n",
    "        dataBuilder = DataBuilder(result.applicationDataModel, collector=result)\n",
    "        for table in tables:\n",
    "            dataBuilder.addTable(table, self.getTable(table))\n",
    "            result.size[table] = self.size[table]\n",
    "        dataBuilder.build();\n",
    "        return result\n",
    "\n",
    "    def getName(self):\n",
    "        \"\"\"\n",
    "        Returns the name of this collector.\n",
    "\n",
    "        :return collector name as a string\n",
    "        \"\"\"\n",
    "        return self.name\n",
    "\n",
    "    def addTables(self, other):\n",
    "        \"\"\"\n",
    "        Adds a set of tables of data from another collector.\n",
    "        An individual table can be set only once.\n",
    "\n",
    "        :param other: another collector\n",
    "        :type other: OPLCollector\n",
    "        :raise ValueError: if the other ADM is empty or if a table name duplicates a name already present in this collector.\n",
    "        \"\"\"\n",
    "\n",
    "        if not other.applicationDataModel:  # is empty\n",
    "            raise ValueError(\"empty collector\")\n",
    "        for tableName in other.applicationDataModel.viewkeys():\n",
    "            if tableName in self.applicationDataModel:\n",
    "                raise ValueError(\"table \" + tableName + \" has already been defined\")\n",
    "        self.applicationDataModel.update(other.applicationDataModel)\n",
    "        self.sparkData.update(other.sparkData)\n",
    "        self.size.update(other.size)\n",
    "        return self\n",
    "\n",
    "    def replaceTable(self, tableName, table, size=None):\n",
    "        \"\"\"\n",
    "        Replaces an individual table of data.\n",
    "\n",
    "        :param tableName:\n",
    "        :type String\n",
    "        :param table:\n",
    "        :type Spark Dataframe\n",
    "        :param size: number of rows in table (None if omitted)\n",
    "        :return: this collector\n",
    "        :raise ValueError: if the table is not already defined in the ADM\n",
    "        \"\"\"\n",
    "        if tableName not in self.applicationDataModel:\n",
    "            raise ValueError(\"table \" + tableName + \"has not been defined\")\n",
    "        self.sparkData[tableName] = table\n",
    "        if size is not None:\n",
    "            self.size[tableName] = size\n",
    "        else:\n",
    "            self.size[tableName] = table.count()\n",
    "        return None\n",
    "\n",
    "    def addData(self, tableName, table, size=None):\n",
    "        \"\"\"\n",
    "        Adds data to an existing table.\n",
    "        Use when a table has several input sources.\n",
    "        Does not deduplicate the data (i.e. allows duplicate rows).\n",
    "\n",
    "        :param tableName:\n",
    "        :type String\n",
    "        :param table:\n",
    "        :type Spark Dataframe\n",
    "        :param size: number of rows in table (None if omitted)\n",
    "        :return: this collector\n",
    "        :raise ValueError: if the table is not already defined in the ADM\n",
    "        \"\"\"\n",
    "        if tableName in self.applicationDataModel:\n",
    "            raise ValueError(\"table \" + tableName + \" has already been defined\")\n",
    "        self.sparkData[tableName] = self.sparkData[tableName].union(table)\n",
    "        count = (self.size[tableName] + size) if (self.size[tableName] is not None and size is not None) else None\n",
    "        self.size[tableName] = count\n",
    "        return self\n",
    "\n",
    "    #NEW\n",
    "    def getADM(self):\n",
    "        \"\"\"\n",
    "        Exposes the application data model for this OPLCollector.\n",
    "        The ADM is represented by a map in which the keys are the table names\n",
    "        and the values are the table schemas held in Spark StructType objects.\n",
    "\n",
    "        :return: the application data model\n",
    "        :rtype: dict<String, StructType>\n",
    "        \"\"\"\n",
    "        return self.applicationDataModel\n",
    "\n",
    "    def setADM(self, applicationDataModel):\n",
    "        \"\"\"\n",
    "        Sets the application data model for this OPLCollector.\n",
    "        The ADM cannot be changed once set.\n",
    "        \"\"\"\n",
    "        if (self.applicationDataModel):  # is not empty or None\n",
    "            raise ValueError(\"ADM has already been defined\")\n",
    "        self.applicationDataModel = applicationDataModel\n",
    "        return self\n",
    "\n",
    "    def getTable(self, tableName):\n",
    "        return self.sparkData[tableName]\n",
    "\n",
    "    def getSchema(self, tableName):\n",
    "        return self.applicationDataModel[tableName]\n",
    "\n",
    "    def selectSchemas(self, *tableNames):\n",
    "        \"\"\"\n",
    "        Returns a subset of the application data model.\n",
    "        \"\"\"\n",
    "        return {tableName: self.applicationDataModel[tableName] for tableName in tableNames}\n",
    "\n",
    "    def selectTables(self, collectorName, *tableNames):\n",
    "        \"\"\"\n",
    "        Creates a new OPLCollector from a subset of the tables in this collector.\n",
    "        The tables in the new collector are copies of the tables in the original.\n",
    "        \"\"\"\n",
    "        adm = self.selectSchemas(tableNames)\n",
    "        data = {tableName: SPARK_SESSION.createDataFrame(self.sparkData[tableName], self.getSchema(tableName))\n",
    "                for tableName in tableNames}\n",
    "        size = {tableName: self.size[tableName] for tableName in tableNames}\n",
    "        return OPLCollector(collectorName, adm, data, size)\n",
    "\n",
    "    def getSize(self, tableName):\n",
    "        \"\"\"\n",
    "        Returns the number of rows in a table.\n",
    "        Note: the Spark data set count method is fairly expensive,\n",
    "        so it is used only if there is no other way to count the number of rows.\n",
    "        It is best to count the rows as the table is being deserialized, as is done in the fromJSON method.\n",
    "        Once counted, the number is stored in the size map for future use.\n",
    "        \"\"\"\n",
    "        if tableName not in self.size:\n",
    "            raise ValueError(\"size not defined for table \" + tableName)\n",
    "        if self.size[tableName] is None:\n",
    "            self.size[tableName] = self.sparkData[tableName].count()\n",
    "        return self.size[tableName]\n",
    "\n",
    "    def buildADM(self):\n",
    "        \"\"\"\n",
    "        Creates the application data model for this collector\n",
    "        \"\"\"\n",
    "        if (self.applicationDataModel):  # is not empty\n",
    "            raise ValueError(\"application data model has already been defined\")\n",
    "        return ADMBuilder(self)\n",
    "\n",
    "    def buildData(self):\n",
    "        \"\"\"\n",
    "        Creates a builder for the data tables for this collector.\n",
    "        Uses this collector's application data model.\n",
    "\n",
    "        :return: a new DataBuilder instance\n",
    "        :raise ValueError: if the application data model has not been defined or if data tables have already been loaded\n",
    "        \"\"\"\n",
    "        if not self.applicationDataModel:  # is empty\n",
    "            raise ValueError(\"application data model has not been defined\")\n",
    "        if self.sparkData:  # is not empty\n",
    "            raise ValueError(\"data tables have already been loaded\")\n",
    "        return DataBuilder(self.applicationDataModel, collector=self)\n",
    "\n",
    "    def setJsonSource(self, source):\n",
    "        \"\"\"\n",
    "        Sets the source for the JSON text that populates the collector.\n",
    "        There is a one-to-one correspondence between an OPLCollector instance and its JSON representation;\n",
    "        that is, the JSON source file must fully include all the data tables to be populated in the collector instance.\n",
    "        Thus, it makes no sense to have more than on JSON source for a collector or to change JSON sources.\n",
    "\n",
    "        :param source: a file-like object containing the JSON text.\n",
    "        :return: this collector instance\n",
    "        :raise ValueError: if JSON source has already been set\n",
    "        \"\"\"\n",
    "        if self.jsonSource is not None:\n",
    "            raise ValueError(\"JSON source has already been set\")\n",
    "        self.jsonSource = source\n",
    "        return self\n",
    "\n",
    "    #REVISED\n",
    "    def fromJSON(self):\n",
    "        \"\"\"\n",
    "        Provides a means to create a collector from JSON.\n",
    "        You must first set the destination (an output stream, file, url, or string) where the JSON will be read.\n",
    "        Then you call the deserializer fromJSON method.\n",
    "        The application data model for the collector must already have been created.\n",
    "\n",
    "        There is a one-to-one correspondence between an OPLCollector instance and its JSON representation;\n",
    "        that is, the JSON source file must fully include all the data tables to be populated in the collector instance.\n",
    "        Methods are provided to merge two collectors with separate JSON sources (addTables),\n",
    "        add a data set to a collector (addTable), and to add data from a data set to an existing table in a collector.\n",
    "\n",
    "        :return: this collector with its data tables filled\n",
    "        :raise ValueError: if the data tables have already been loaded\n",
    "        \"\"\"\n",
    "        if self.sparkData:  # is not empty\n",
    "            raise ValueError(\"data tables have already been loaded\")\n",
    "        # data: dict {tableName_0: [{fieldName_0: fieldValue_0, ...}, ...], ...}\n",
    "        data = json.load(self.jsonSource)\n",
    "        builder = self.buildData()\n",
    "        for tableName, tableData in data.viewitems():\n",
    "            count = len(tableData)\n",
    "            tableRows = (Row(**fields) for fields in tableData)\n",
    "            builder = builder.addTable(tableName,\n",
    "                                       SPARK_SESSION.createDataFrame(tableRows, self.getADM()[tableName]),\n",
    "                                       count)   # would like to count the rows as they are read instead,\n",
    "                                                # but don't see how\n",
    "        builder.build()\n",
    "        return self\n",
    "\n",
    "    def setJsonDestination(self, destination):\n",
    "        \"\"\"\n",
    "        Sets the destination for the JSON serialization.\n",
    "        Replaces an existing destination if one has been set previously.\n",
    "\n",
    "        :param destination: an output string, stream, file, or URL\n",
    "        :return: this collector\n",
    "        \"\"\"\n",
    "        self.jsonDestination = destination\n",
    "        return self\n",
    "\n",
    "    #REVISED\n",
    "    def toJSON(self):\n",
    "        \"\"\"\n",
    "        Provides a means to write the application data as JSON.\n",
    "        You must first set the destination (an output stream, file, url, or string) where the JSON will be written.\n",
    "        Then you call the serializer toJSON method.\n",
    "        \"\"\"\n",
    "        self.jsonDestination.write(\"{\\n\")                           # start collector object\n",
    "        firstTable = True\n",
    "        for tableName in self.sparkData:\n",
    "            if not firstTable:\n",
    "                self.jsonDestination.write(',\\n')\n",
    "            else:\n",
    "                firstTable = False\n",
    "            self.jsonDestination.write('\"' + tableName + '\" : [\\n') # start table list\n",
    "            firstRow = True\n",
    "            for row in self.sparkData[tableName].toJSON().collect():# better to use toLocalIterator() but it gives a timeout error\n",
    "                if not firstRow:\n",
    "                    self.jsonDestination.write(\",\\n\")\n",
    "                else:\n",
    "                    firstRow= False\n",
    "                self.jsonDestination.write(row)                     # write row object\n",
    "            \n",
    "            self.jsonDestination.write(\"\\n]\")                       # end table list\n",
    "        self.jsonDestination.write(\"\\n}\")                           # end collector object\n",
    "\n",
    "    #REVISED\n",
    "    def displayTable(self, tableName, out=sys.stdout):\n",
    "        \"\"\"\n",
    "        Prints the contents of a table.\n",
    "\n",
    "        :param out: a file or other print destination where the table will be written\n",
    "        \"\"\"\n",
    "        out.write(\"collector: \" + self.getName() + \"\\n\")\n",
    "        out.write(\"table: \" + tableName + \"\\n\")\n",
    "        self.getTable(tableName).show(self.getSize(tableName), truncate=False)\n",
    "\n",
    "    # REVISED\n",
    "    def display(self, out=sys.stdout):\n",
    "        \"\"\"\n",
    "        Prints the contents of all tables in this collector.\n",
    "\n",
    "        :param out: a file or other print destination where the tables will be written\n",
    "        \"\"\"\n",
    "        for tableName in self.sparkData:\n",
    "            self.displayTable(tableName, out=out)\n",
    "\n",
    "\n",
    "# end class OPLCollector\n",
    "\n",
    "def getFromObjectStorage(credentials, container=None, filename=None):\n",
    "    \"\"\"\n",
    "    Returns a stream containing a file's content from Bluemix Object Storage.\n",
    "\n",
    "    :param credentials a dict generated by the Insert to Code  service of the host Notebook\n",
    "    :param container the name of the container as specified in the credentials (defaults to the credentials entry)\n",
    "    :param filename the name of the file to be accessed (note: if there is more than one file in the container,\n",
    "    you might prefer to enter the names directly; otherwise, defaults to the credentials entry)\n",
    "    \"\"\"\n",
    "\n",
    "    if not container:\n",
    "        container = credentials['container']\n",
    "    if not filename:\n",
    "        filename = credentials['filename']\n",
    "\n",
    "    url1 = ''.join([credentials['auth_url'], '/v3/auth/tokens'])\n",
    "    data = {'auth': {'identity': {'methods': ['password'],\n",
    "                                  'password': {\n",
    "                                      'user': {'name': credentials['username'], 'domain': {'id': credentials['domain_id']},\n",
    "                                               'password': credentials['password']}}}}}\n",
    "    headers1 = {'Content-Type': 'application/json'}\n",
    "    resp1 = requests.post(url=url1, data=json.dumps(data), headers=headers1)\n",
    "    resp1_body = resp1.json()\n",
    "    for e1 in resp1_body['token']['catalog']:\n",
    "        if (e1['type'] == 'object-store'):\n",
    "            for e2 in e1['endpoints']:\n",
    "                if (e2['interface'] == 'public' and e2['region'] == credentials['region']):\n",
    "                    url2 = ''.join([e2['url'], '/', container, '/', filename])\n",
    "    s_subject_token = resp1.headers['x-subject-token']\n",
    "    headers2 = {'X-Auth-Token': s_subject_token, 'accept': 'application/json'}\n",
    "    resp2 = requests.get(url=url2, headers=headers2, stream=True)\n",
    "    return resp2.raw\n",
    "\n",
    "\n",
    "class DataBuilder(object):\n",
    "    \"\"\"\n",
    "        Builds the Spark datasets to hold the application data.\n",
    "        Used when the data are created programmatically.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, applicationDataModel, collector=None):\n",
    "        \"\"\"\n",
    "        Creates a builder for loading the Spark datasets.\n",
    "\n",
    "        :param applicationDataModel\n",
    "        :param collector: if present, loads the data tables and their sizes directly into the collector;\n",
    "        if not present or null, the Spark data dict is returned directly\n",
    "        :return: a new DataBuilder instance\n",
    "        :raise ValueError: if the application data model has not been defined\n",
    "        \"\"\"\n",
    "        if not applicationDataModel:  # is empty\n",
    "            raise ValueError(\"application data model has not been defined\")\n",
    "        self.applicationDataModel = applicationDataModel\n",
    "        self.collector = collector\n",
    "        self.result = {}\n",
    "        self.length = {}\n",
    "\n",
    "    def addTable(self, tableName, data, size=None):\n",
    "        \"\"\"\n",
    "        Get the external data and create the corresponding application dataset.\n",
    "        Assumes that the schema of this table is already present in the ADM.\n",
    "\n",
    "        :param data: a Spark dataset\n",
    "        :param size: length number of rows in table (null if omitted)\n",
    "        :return this builder instance\n",
    "        :raise ValueError: if the table is not included in the ADM or if the table has already been loaded\n",
    "        \"\"\"\n",
    "        if tableName not in self.applicationDataModel:\n",
    "            raise ValueError(\"table \" + tableName + \"has not been defined\")\n",
    "        if tableName in self.result:\n",
    "            raise ValueError(\"table \" + tableName + \"has already been loaded\")\n",
    "        self.result[tableName] = data\n",
    "        self.length[tableName] = size\n",
    "        return self\n",
    "\n",
    "    def copyTable(self, tableName, data, size=None):\n",
    "        return self.addTable(tableName,\n",
    "                             SPARK_SESSION.createDataFrame(data.rdd()), size);\n",
    "\n",
    "    def addEmptyTable(self, tableName):\n",
    "        return self.addTable(tableName,\n",
    "                             SPARK_SESSION.createDataFrame(SPARK_CONTEXT.emptyRDD(),\n",
    "                                                           self.applicationDataModel[tableName]), 0)\n",
    "\n",
    "    #NEW\n",
    "    def referenceTable(self, tableName):\n",
    "        \"\"\"\n",
    "        Enables referring to a table in the collector under construction to create a new table.\n",
    "        Can be used in SQL statements. \n",
    "        \n",
    "        :param tableName: \n",
    "        :type tableName: String\n",
    "        :return: \n",
    "        :rtype:\n",
    "        \"\"\"\n",
    "        if tableName not in self.result:\n",
    "            raise ValueError(tableName + \" does not exist\")\n",
    "        return self.result.get(tableName)\n",
    "\n",
    "    def build(self):\n",
    "        \"\"\"\n",
    "        Completes building the Spark data.\n",
    "        Registers the application data sets as Spark SQL tables.\n",
    "        If an OPLCollector has been supplied in the constructor, loads the data tables and their sizes into it.\n",
    "\n",
    "        :return a dict of table names to Spark data sets containing the application data\n",
    "        :raise ValueError:  if a table in the ADM has no associated data or if data tables have already been loaded into the collector\n",
    "        \"\"\"\n",
    "        for tableName in self.applicationDataModel:\n",
    "            if tableName not in self.result:\n",
    "                raise ValueError(\"table \" + tableName + \"has no data\")\n",
    "        for tableName in self.result:\n",
    "            self.result[tableName].createOrReplaceTempView(tableName)\n",
    "        if self.collector is not None:\n",
    "            if self.collector.sparkData:  # is not empty\n",
    "                raise ValueError(\"data tables have already been loaded\")\n",
    "            self.collector.sparkData = self.result\n",
    "            self.collector.size = self.length\n",
    "        return self.result\n",
    "\n",
    "    def retrieveSize(self):\n",
    "        \"\"\"\n",
    "        :return the size dict created by this builder\n",
    "        Note: calling this method before the build method could return an inaccurate result\n",
    "        \"\"\"\n",
    "        return self.length\n",
    "\n",
    "\n",
    "# end class DataBuilder\n",
    "\n",
    "class ADMBuilder(object):\n",
    "    \"\"\"\n",
    "    Builds an Application Data Model that associates a set of Spark Datasets with their schemas.\n",
    "    Usage:\n",
    "\n",
    "    adm= ADMBuilder()\\\n",
    "        .addSchema(\"warehouse\", buildSchema(\n",
    "            (\"location\", StringType()),\n",
    "            (\"capacity\", DoubleType()))\\\n",
    "        .addSchema(\"route\", buildSchema(\n",
    "            (\"from\", StringType()),\n",
    "            (\"to\", StringType()),\n",
    "            (\"capacity\", DoubleType()))\\\n",
    "        .build();\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, collector=None):\n",
    "        \"\"\"\n",
    "        Creates a new builder.\n",
    "        :param collector if present, loads the application data model directly into the collector;\n",
    "        if not present or null, the ADM map is returned directly\n",
    "        \"\"\"\n",
    "        self.collector = collector\n",
    "        self.result = {}\n",
    "\n",
    "    def addSchema(self, tableName, tupleSchema):\n",
    "        \"\"\"\n",
    "        Adds a new table schema to the ADM.\n",
    "\n",
    "        :param tupleSchema can be built with the buildSchema function\n",
    "        :return this builder\n",
    "        :raise ValueError: if a schema for tableName has already been defined\n",
    "        \"\"\"\n",
    "        if tableName in self.result:\n",
    "            raise ValueError(\"tuple schema \" + tableName + \" has already been defined\")\n",
    "        self.result[tableName] = tupleSchema\n",
    "        return self\n",
    "\n",
    "    #NEW\n",
    "    def referenceSchema(self, tableName):\n",
    "        \"\"\"\n",
    "        Enables referring to a schema in the ADM under construction to create a  new schema.\n",
    "\n",
    "        :param tableName\n",
    "        :return the schema\n",
    "        \"\"\"\n",
    "        if tableName not in self.result:\n",
    "            raise ValueError(\"tuple schema \" + tableName + \" does not exist\")\n",
    "        return self.result[tableName]\n",
    "\n",
    "    def build(self):\n",
    "        \"\"\"\n",
    "        Completes building the application data model.\n",
    "        If an OPLCollector has been supplied in the constructor, loads the ADM into it.\n",
    "\n",
    "        :return the ADM\n",
    "        :raise ValueError: if the ADM for the collector has already been defined\n",
    "        \"\"\"\n",
    "        if self.collector is not None:\n",
    "            if self.collector.applicationDataModel:  # is not empty\n",
    "                raise ValueError(\"application data model has already been defined\")\n",
    "            self.collector.applicationDataModel = self.result\n",
    "        return self.result\n",
    "\n",
    "# end class ADMBuilder\n",
    "\n",
    "def buildSchema(*fields):\n",
    "    \"\"\"\n",
    "    Creates a schema from a list field tuples\n",
    "    The resulting schema is an instance of a Spark StructType.\n",
    "    :param fields:\n",
    "    :type fields: tuple<String, DataType>\n",
    "    :return:\n",
    "    :rtype: StructType\n",
    "    \"\"\"\n",
    "    schema = StructType()\n",
    "    for fieldName, fieldType in fields:\n",
    "        schema = schema.add(fieldName, fieldType, False, None)\n",
    "    return schema\n",
    "# end buildSchema\n",
    "\n",
    "#NEW\n",
    "class SchemaBuilder:\n",
    "    \"\"\"\n",
    "    Builds a tuple schema.\n",
    "    Strictly speaking, this builder is not needed since the StructType class provides the necessary functionality.\n",
    "    However, it is provided as a convenience.\n",
    "    Only the following data types are supported in the schema: String, integer, float (represented as Double), and 1-dimensional arrays of integer or float.\n",
    "    The array types are supported only for internal use and cannot be serialized to or deserialized from JSON.\n",
    "    Note, the fields in the resulting schema are sorted in dictionary order by name to insure correct matching with data elements.\n",
    "\n",
    "    Usage:\n",
    "        StructType warehouseSchema= (new OPLTuple.SchemaBuilder()).addField(\"location\", DataTypes.StringType).addField(\"capacity\", DataTypes.DoubleType).buildSchema();\n",
    "    The fields of the resulting StructType are not nullable and have no metadata.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self):\n",
    "        # fields is a dictionary<String, StructField>\n",
    "        self.fields= {}\n",
    "\n",
    "    def addField(self, fieldName, fieldType):\n",
    "        if self.fields.has_key(fieldName):\n",
    "            raise ValueError(\"field \" + fieldName + \" has already been set\")\n",
    "        self.fields[fieldName]= StructField(fieldName, fieldType, False)\n",
    "        return self;\n",
    "\n",
    "    def addFields(self, *fields):\n",
    "        \"\"\"\n",
    "        Adds fields from a list field tuples\n",
    "        :param fields: tuple<String, DataType>\n",
    "        :return: StructType\n",
    "        \"\"\"\n",
    "        for field in fields:\n",
    "            fieldName, fieldType= field\n",
    "            self.addField(fieldName, fieldType)\n",
    "        return self\n",
    "\n",
    "    def copyField(self, otherSchema, fieldName):\n",
    "        \"\"\"\n",
    "        Copies fields from another schema\n",
    "        :param otherSchema: StructType\n",
    "        :param fieldName: String\n",
    "        :return: StructType\n",
    "        \"\"\"\n",
    "        if self.fields.has_key(fieldName):\n",
    "            raise ValueError(\"field \" + fieldName + \" has already been set\")\n",
    "        self.fields[fieldName]= otherSchema[fieldName]\n",
    "        return self\n",
    "\n",
    "    def copyFields(self, otherSchema):\n",
    "        for fieldName in otherSchema.names:\n",
    "            self.copyField(otherSchema, fieldName)\n",
    "        return self\n",
    "\n",
    "    def buildSchema(self):\n",
    "        return StructType(self.fields.values())\n",
    "# end class SchemaBuilder"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.4 The Spark Data Model for the Warehouse Location Application"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the tools in the OPLCollector class, the application data model (ADM) is defined in three collector objects, each of which has a corresponding representation in OPL and an associated JSON data file. The ADM is defined as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "networkDataModel = ADMBuilder()\\\n",
    "    .addSchema(\"warehouses\", buildSchema(\n",
    "        (\"location\",     StringType()),\n",
    "        (\"fixedCost\",    DoubleType()),\n",
    "        (\"capacityCost\", DoubleType())))\\\n",
    "    .addSchema(\"routes\", buildSchema(\n",
    "        (\"location\",     StringType()),\n",
    "        (\"store\",        StringType()),\n",
    "        (\"shippingCost\", DoubleType())))\\\n",
    "    .addSchema(\"stores\", buildSchema(\n",
    "        (\"storeId\",      StringType())))\\\n",
    "    .addSchema(\"mapCoordinates\", buildSchema(\n",
    "        (\"location\",     StringType()),\n",
    "        (\"lon\",          DoubleType()),\n",
    "        (\"lat\",          DoubleType())))\\\n",
    "    .build()\n",
    "\n",
    "demandDataModel = ADMBuilder()\\\n",
    "    .addSchema(\"demands\", buildSchema(\n",
    "        (\"store\",        StringType()),\n",
    "        (\"scenarioId\",   StringType()),\n",
    "        (\"amount\",       DoubleType())))\\\n",
    "    .addSchema(\"scenarios\", buildSchema(\n",
    "        (\"id\",           StringType()),\n",
    "        (\"totalDemand\",  DoubleType()),\n",
    "        (\"periods\",      DoubleType())))\\\n",
    "    .build()\n",
    "\n",
    "warehousingResultDataModel= ADMBuilder()\\\n",
    "    .addSchema(\"objectives\", buildSchema(\n",
    "        (\"problem\",     StringType()),\n",
    "        (\"dExpr\",       StringType()),\n",
    "        (\"scenarioId\",  StringType()),\n",
    "        (\"iteration\",   IntegerType()),\n",
    "        (\"value\",       DoubleType())))\\\n",
    "    .addSchema(\"openWarehouses\", buildSchema(\n",
    "        (\"location\",    StringType()),\n",
    "        (\"scenarioId\",  StringType()),\n",
    "        (\"iteration\",   IntegerType()),\n",
    "        (\"open\",        IntegerType()),\n",
    "        (\"capacity\",    DoubleType())))\\\n",
    "    .addSchema(\"shipments\", buildSchema(\n",
    "        (\"location\",    StringType()),\n",
    "        (\"store\",       StringType()),\n",
    "        (\"scenarioId\",  StringType()),\n",
    "        (\"iteration\",   IntegerType()),\n",
    "        (\"amount\",      DoubleType())))\\\n",
    "    .build()\n",
    "    \n",
    "# Note: the \"MapCoordinates table and the \"scenarioId\" and \"iteration\" fields are not used in this notebook but are included for use in other contexts."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.5 The Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The data for the warehouse location problem consist of two JSON files. The first contains the characteristics of the distribution network, the potential warehouse locations and their capital costs, the store locations, and the transportation routes between the stores and the warehouses and their shipping costs. The second contains the demands at each store. These two constitute all the data required to optimize the warehouse network.\n",
    "\n",
    "- Warehousing-data.json contains the warehouses, stores, routes, and mapCoordinates\n",
    "- Warehousing-sales_data-nominal_scenario.json contains the scenarios and demands\n",
    "\n",
    "These two data files resides in the DSX Community at: __[Potential warehouse locations](https://apsportal.ibm.com/exchange/public/entry/view/2a493fe2f0d475f0b5b52bce6191f129)__ and __[Demand per store](https://apsportal.ibm.com/exchange/public/entry/view/0a5f75c8e2177f0f64fe22e677588b1a)__. \n",
    "\n",
    "Extracts of these two files are shown in the notebook *Locating Warehouses to Minimize Costs Case 1*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the <code>OPLCollector</code> class and its adjuncts, the following cell reads the input data and creates the Spark datasets with which to populate the decision model: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "collector: warehousingData\n",
      "table: warehouses\n",
      "+------------------+---------+------------+\n",
      "|location          |fixedCost|capacityCost|\n",
      "+------------------+---------+------------+\n",
      "|Brockton, MA      |550000.0 |148.0       |\n",
      "|Bristol, CT       |600000.0 |148.0       |\n",
      "|Union City, NJ    |600000.0 |148.0       |\n",
      "|New York, NY      |500000.0 |148.0       |\n",
      "|Philadelphia, PA  |500000.0 |148.0       |\n",
      "|Parkville, MD     |550000.0 |148.0       |\n",
      "|Greensboro, NC    |500000.0 |148.0       |\n",
      "|Goose Creek, SC   |500000.0 |148.0       |\n",
      "|Lawrenceville, GA |450000.0 |148.0       |\n",
      "|Jacksonville, FL  |550000.0 |148.0       |\n",
      "|Birmingham, AL    |450000.0 |148.0       |\n",
      "|Memphis, TN       |450000.0 |148.0       |\n",
      "|Frankfort, KY     |500000.0 |148.0       |\n",
      "|Akron, OH         |500000.0 |148.0       |\n",
      "|Dayton, OH        |500000.0 |148.0       |\n",
      "|West Lafayette, IN|500000.0 |148.0       |\n",
      "|Taylor, MI        |500000.0 |148.0       |\n",
      "|Dubuque, IA       |400000.0 |148.0       |\n",
      "|Beloit, WI        |500000.0 |148.0       |\n",
      "|Chicago, IL       |500000.0 |148.0       |\n",
      "|Saint Peters, MO  |500000.0 |148.0       |\n",
      "|Broken Arrow, OK  |500000.0 |148.0       |\n",
      "|Dallas, TX        |450000.0 |148.0       |\n",
      "|Denver, CO        |500000.0 |148.0       |\n",
      "|Phoenix, AZ       |500000.0 |148.0       |\n",
      "|Los Angeles, CA   |650000.0 |148.0       |\n",
      "|San Francisco, CA |650000.0 |148.0       |\n",
      "|Salem, OR         |500000.0 |148.0       |\n",
      "+------------------+---------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import cStringIO\n",
    "networkDataSource = cStringIO.StringIO(requests.get(\"https://apsportal.ibm.com/exchange-api/v1/entries/2a493fe2f0d475f0b5b52bce6191f129/data?accessKey=4d9d8f5f8569ea123076f7ae988649f4\").text) # Warehousing-data.json\n",
    "demandDataSource =  cStringIO.StringIO(requests.get(\"https://apsportal.ibm.com/exchange-api/v1/entries/0a5f75c8e2177f0f64fe22e677588b1a/data?accessKey=e3e48701299acea80483b1624d696795\").text) # Warehousing-sales_data-nominal_scenario.json\n",
    "\n",
    "warehousingData= OPLCollector(\"warehousingData\", networkDataModel).setJsonSource(networkDataSource).fromJSON()\n",
    "warehousingData.addTables(OPLCollector(\"demandData\", demandDataModel).setJsonSource(demandDataSource).fromJSON())\n",
    "\n",
    "warehousingData.displayTable(\"warehouses\", sys.stdout)\n",
    "\n",
    "networkDataSource.close()\n",
    "demandDataSource.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Uncomment the following statement and rerun the cell to see the data. Warning: it creates a lengthy table.\n",
    "#warehousingData.displayTable(\"stores\", sys.stdout)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.6 Optimization Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here is the OPL statement of the Warehousing optimization model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "warehousing_inputs='''\n",
    " //Input Data\n",
    "  \n",
    " {Warehouse} warehouses= ...;\t//Denotes reading from a data source\n",
    " \n",
    " {Store} stores= ...;\n",
    " \n",
    " {Route} routes= ...;\n",
    " \n",
    " {Demand} demands= ...;\n",
    " float demand[routes]= [r: d.amount | r in routes,  d in demands: r.store==d.store]; //demand at the store at the end of route r\n",
    " \n",
    " {Scenario} scenarios= ...;\n",
    " Scenario scenario= first(scenarios); //scenarios is a singleton set\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "warehousing_dotmod='''\n",
    " dvar boolean open[warehouses];\n",
    " dvar float+ capacity[warehouses];\t\t//pallets\n",
    " dvar float+ ship[routes] in 0.0..1.0;\t//percentage of each store's demand shipped on each route\n",
    " \n",
    " dexpr float capitalCost=\tsum(w in warehouses) (w.fixedCost*open[w] + w.capacityCost*capacity[w]);\n",
    " dexpr float operatingCost=\tsum(r in routes) r.shippingCost*demand[r]*ship[r];\n",
    " dexpr float totalCost=\t\tsum(w in warehouses) (w.fixedCost*open[w] + w.capacityCost*capacity[w]) +\n",
    " \t\t\t\t\t\t\tsum(r in routes) r.shippingCost*demand[r]*ship[r];\n",
    " \n",
    " constraint ctCapacity[warehouses];\n",
    " constraint ctDemand[stores];\n",
    " constraint ctSupply[routes];\n",
    " \n",
    " minimize totalCost;\t\t\t\t\t// $/yr\n",
    " subject to {\n",
    " \t \n",
    " \tforall(w in warehouses)\n",
    "//\t  Cannot ship more out of a warehouse than its capacity\n",
    " \t  ctCapacity[w]: capacity[w] >= sum(r in routes: r.location==w.location) demand[r]*ship[r];\n",
    " \t \n",
    "\tforall(s in stores)\n",
    "//    Must ship at least 100% of each store's demand\n",
    "\t  ctDemand[s]: sum(r in routes: r.store==s.storeId) ship[r] >= 1.0;\n",
    "   \t   \n",
    "\tforall(r in routes, w in warehouses: w.location==r.location)\n",
    "//\t  Can only ship along a supply route if its warehouse is open\t  \n",
    "\t  ctSupply[r]: -ship[r] >= -open[w];\t//ship[r] <= open[w]\n",
    "   \n",
    " }\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "warehousing_outputs= '''\n",
    " //Output Data\n",
    " \n",
    " {Objective} objectives= {\n",
    "    <\"Warehousing\", \"capitalCost\", scenario.id, 0, capitalCost>,\n",
    "    <\"Warehousing\", \"operatingCost\", scenario.id, 0, operatingCost>, \n",
    "    <\"Warehousing\", \"totalCost\", scenario.id, 0, totalCost>};\n",
    " \n",
    " {Shipment} shipments= {<r.location, r.store, scenario.id, 0, ship[r]*d.amount> | r in routes, d in demands: r.store==d.store && ship[r]>0.0};\n",
    " \n",
    " {OpenWarehouse} openWarehouses= {<w.location, scenario.id, 0, open[w], capacity[w]> | w in warehouses};\n",
    "'''"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.7 Solving the Warehousing Model with IBM Decision Optimization on Cloud"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.7.1 Get Your Credentials for IBM Decision Optimization on Cloud"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to use IBM Decision Optimization on Cloud, you need to insert your credentials here (if you don't already have them, you can register for a trial  __[here](https://dropsolve-oaas.docloud.ibmcloud.com/software/analytics/docloud)__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "url= \"\" # ENTER YOUR URL HERE\n",
    "key= \"\" # ENTER YOUR KEY HERE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.7.2 The Optimizer Class"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The optimization computation uses IBM Decision Optimization on Cloud, which exposes IBM's CPLEX through a cloud-based interface. In order to simplify applications built on this cloud platform, the calls to the solver have been abstracted as the Optimizer class, shown below. This class is independent of the actual decision model and instance data, and so it can be reused in other decision optimization applications without modification. Here is the Python code for Optimizer class and some related functions (go to edit mode to see the contents of this hidden cell)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# @hidden_cell\n",
    "'''\n",
    "Created on Feb 9, 2017\n",
    "\n",
    "@author: bloomj\n",
    "'''\n",
    "try:\n",
    "    import docloud\n",
    "except:\n",
    "    if hasattr(sys, 'real_prefix'):\n",
    "        #we are in a virtual env.\n",
    "        !pip install docloud \n",
    "    else:\n",
    "        !pip install --user docloud\n",
    "\n",
    "from docloud.job import JobClient\n",
    "from docloud.status import JobSolveStatus, JobExecutionStatus\n",
    "\n",
    "from urlparse import urlparse\n",
    "\n",
    "import fileinput\n",
    "import urllib\n",
    "import cStringIO\n",
    "from pprint import pprint\n",
    "\n",
    "class Optimizer(object):\n",
    "    '''\n",
    "     Handles the actual optimization task.\n",
    "     Creates and executes a job builder for an optimization problem instance.\n",
    "     Encapsulates the DOCloud API.\n",
    "     This class is designed to facilitate multiple calls to the optimizer, such as would occur in a decomposition algorithm,\n",
    "     although it transparently supports single use as well.\n",
    "     In particular, the data can be factored into a constant data set that does not vary from run to run (represented by a JSON or .dat file)\n",
    "     and a variable piece that does vary (represented by a Collector object).\n",
    "     The optimization model can also be factored into two pieces, a best practice for large models and multi-models:\n",
    "     A data model that defines the tuples and tuple sets that will contain the input and output data.\n",
    "     An optimization model that defines the decision variables, decision expressions, objective function, \n",
    "     constraints, and pre- and post-processing data transformations.\n",
    "     Factoring either the data or the optimization model in this fashion is optional.\n",
    "     \n",
    "     The problem instance is specified by the OPL model and input data received from the invoking (e.g. ColumnGeneration) instance.\n",
    "     Input and output data are realized as instances of OPLCollector, which in turn are specified by their respective schemas.\n",
    "     This class is completely independent of the specific optimization problem to be solved.\n",
    "    '''\n",
    "\n",
    "    def __init__(self, problemName, model=None, resultDataModel=None, credentials=None, *attachments):\n",
    "        '''\n",
    "         Constructs an Optimizer instance.\n",
    "         The instance requires an optimization model as a parameter.\n",
    "         You can also provide one or more data files as attachments, either in OPL .dat or in JSON format. This data does not\n",
    "         change from solve to solve. If you have input data that does change, you can provide it to the solve method as an OPLCollector object.\n",
    "         :param problemName: name of this optimization problem instance\n",
    "         :type problemName: String    \n",
    "         :param model: an optimization model written in OPL\n",
    "         :type model: Model.Source object or String\n",
    "         :param resultDataModel: the application data model for the results of the optimization\n",
    "         :type resultDataModel: dict<String, StructType>\n",
    "         :param credentials: DOcplexcloud url and api key\n",
    "         :type credentials: {\"url\":String, \"key\":String}\n",
    "         :param attachments: URLs for files representing the data that does not vary from solve to solve\n",
    "         :type attachments: list<URL>\n",
    "        '''\n",
    "        self.name= problemName\n",
    "        self.model= model\n",
    "        self.resultDataModel= resultDataModel\n",
    "        self.attachData(attachments)\n",
    "        self.streamsRegistry= []\n",
    "        self.history= []\n",
    "        \n",
    "        self.credentials= credentials\n",
    " \n",
    "        self.jobclient= JobClient(credentials[\"url\"], credentials[\"key\"]);\n",
    "        self.solveStatus= JobSolveStatus.UNKNOWN;\n",
    "        \n",
    "    def getName(self):\n",
    "        \"\"\"\n",
    "        Returns the name of this problem\n",
    "        \"\"\"\n",
    "        return self.name\n",
    "    \n",
    "    def setOPLModel(self, name, dotMods=None, modelText=None):\n",
    "        '''\n",
    "         Sets the OPL model.\n",
    "         This method can take any number of dotMod arguments, but\n",
    "         there are two common use cases:\n",
    "         First, the optimization model can be composed of two pieces: \n",
    "             A data model that defines the tuples and tuple sets that will contain the input and output data.\n",
    "             An optimization model that defines the decision variables, decision expressions, objective function, \n",
    "             constraints, and pre- and post-processing data transformations.\n",
    "             The two are concatenated, so they must be presented in that order.\n",
    "             If such a composite model is used, you do not need to import the data model into the optimization model using an OPL include statement.\n",
    "         Second, you do not have to use a separate data model, in which case a single dotMod must be provided \n",
    "         which encompasses both the data model and the optimization model.  \n",
    "        @param name: the name assigned to this OPL model (should have the format of a file name with a .mod extension)\n",
    "        @type name: String\n",
    "        @param dotMods: URLs pointing to OPL .mod files, which will be concatenated in the order given\n",
    "        @type dotMods: List<URL>\n",
    "        @param modelText: the text of the OPL model, which will be concatenated in the order given\n",
    "        @type modelText: List<String>\n",
    "        @return this optimizer\n",
    "        @raise ValueError if a model has already been defined or if dotMods or modelText is empty\n",
    "        '''\n",
    "        if self.model is not None:\n",
    "            raise ValueError(\"model has already been set\")\n",
    "        self.model= ModelSource(name=name, dotMods=dotMods, modelText=modelText)\n",
    "        return self\n",
    "    \n",
    "    def setResultDataModel(self, resultDataModel):\n",
    "        '''\n",
    "        Sets the application data model for the results of the optimization\n",
    "        @param resultDataModel: the application data model for the results of the optimization\n",
    "        @type resultDataModel: dict<String, StructType>\n",
    "        '''\n",
    "        if self.resultDataModel is not None:\n",
    "            raise ValueError(\"results data model has already been defined\")        \n",
    "        self.resultDataModel = resultDataModel\n",
    "        return self\n",
    "    \n",
    "    def attachData(self, attachments):\n",
    "        '''\n",
    "        Attaches one or more data files, either in OPL .dat or in JSON format. This data does not\n",
    "        change from solve to solve. If you have input data that does change, you can provide it as a Collector object.\n",
    "        @param attachments: files representing the data that does not vary from solve to solve\n",
    "        @type attachments: list<URL>\n",
    "        @return this optimizer\n",
    "        @raise ValueError if an item of the same name has already been attached\n",
    "        '''\n",
    "        self.attachments= {}\n",
    "        if attachments is not None:\n",
    "            for f in attachments:\n",
    "                fileName= os.path.splitext(os.path.basename(urlparse(f)))[0]\n",
    "                if fileName in self.attachments:\n",
    "                    raise ValueError(fileName+ \" already attached\")\n",
    "                self.attachments[fileName]= f\n",
    "        return self;\n",
    "    \n",
    "    def solve(self, inputData=None, solutionId=\"\"):\n",
    "        '''\n",
    "        Solves an optimization problem instance by calling the DOCloud solve service (Oaas).\n",
    "        Creates a new job request, incorporating any changes to the variable input data, \n",
    "        for a problem instance to be processed by the solve service. \n",
    "        Once the problem is solved, the results are mapped to an instance of an OPL Collector.\n",
    "        Note: this method will set a new destination for the JSON serialization of the input data.\n",
    "        @param inputData: the variable, solve-specific input data\n",
    "        @type inputData: OPLCollector\n",
    "        @param solutionId: an identifier for the solution, used in iterative algorithms (set to empty string if not needed)\n",
    "        @type solutionId: String\n",
    "        @return: a solution collector\n",
    "        '''\n",
    "        inputs= []\n",
    "        if self.model is None:\n",
    "            raise ValueError(\"A model attachment must be provided to the optimizer\")\n",
    "        if self.model: #is not empty\n",
    "            stream= self.model.toStream()\n",
    "            inputs.append({\"name\":self.model.getName(), \"file\":stream})\n",
    "            self.streamsRegistry.append(stream)\n",
    "        if self.attachments: #is not empty\n",
    "            for f in self.attachments:\n",
    "                stream= urllib.FancyURLopener(self.attachments[f])\n",
    "                inputs.append({\"name\":f, \"file\":stream})\n",
    "                self.streamsRegistry.append(stream)\n",
    "        if inputData is not None:\n",
    "            outStream = cStringIO.StringIO()\n",
    "            inputData.setJsonDestination(outStream).toJSON()\n",
    "            inStream = cStringIO.StringIO(outStream.getvalue())\n",
    "            inputs.append({\"name\": inputData.getName()+\".json\", \"file\": inStream})\n",
    "            self.streamsRegistry.extend([outStream, inStream])\n",
    "       \n",
    "        response= self.jobclient.execute(\n",
    "            input= inputs, \n",
    "            output= \"results.json\", \n",
    "            load_solution= True, \n",
    "            log= \"solver.log\", \n",
    "            gzip= True,\n",
    "            waittime= 300,  #seconds\n",
    "            delete_on_completion= False)\n",
    "         \n",
    "        self.jobid= response.jobid\n",
    "        \n",
    "        status= self.jobclient.get_execution_status(self.jobid)\n",
    "        if status==JobExecutionStatus.PROCESSED:\n",
    "            results= cStringIO.StringIO(response.solution)\n",
    "            self.streamsRegistry.append(results)\n",
    "            self.solveStatus= response.job_info.get('solveStatus') #INFEASIBLE_SOLUTION or UNBOUNDED_SOLUTION or OPTIMAL_SOLUTION or...\n",
    "            solution= (OPLCollector(self.getName()+\"Result\"+solutionId, self.resultDataModel)).setJsonSource(results).fromJSON()\n",
    "            self.history.append(solution)\n",
    "        elif status==JobExecutionStatus.FAILED:\n",
    "            # get failure message if defined\n",
    "            message= \"\"\n",
    "            if (response.getJob().getFailureInfo() != None):\n",
    "                message= response.getJob().getFailureInfo().getMessage()\n",
    "            print(\"Failed \" +message)\n",
    "        else:\n",
    "            print(\"Job Status: \" +status)\n",
    "        \n",
    "        for s in self.streamsRegistry:\n",
    "            s.close();\n",
    "        self.jobclient.delete_job(self.jobid);\n",
    "        \n",
    "        return solution\n",
    "    \n",
    "    def getSolveStatus(self):\n",
    "        \"\"\"\n",
    "        @return the solve status as a string\n",
    "        Attributes:\n",
    "            UNKNOWN: The algorithm has no information about the solution.\n",
    "            FEASIBLE_SOLUTION: The algorithm found a feasible solution.\n",
    "            OPTIMAL_SOLUTION: The algorithm found an optimal solution.\n",
    "            INFEASIBLE_SOLUTION: The algorithm proved that the model is infeasible.\n",
    "            UNBOUNDED_SOLUTION: The algorithm proved the model unbounded.\n",
    "            INFEASIBLE_OR_UNBOUNDED_SOLUTION: The model is infeasible or unbounded.\n",
    "        \"\"\"\n",
    "        return self.solveStatus\n",
    "    \n",
    "# end class Optimizer        \n",
    "    \n",
    "class ModelSource(object):\n",
    "    '''\n",
    "     This class manages the OPL source code for an optimization model.\n",
    "     It can use an OPL model specified by one or more files, indicated by their URLs, or\n",
    "     it can use an OPL model specified by one or more Strings. \n",
    "     Use of one OPL component is the norm, but this class also\n",
    "     enables factoring an OPL model into a data model and an optimization model.\n",
    "     Using such a two-piece factorization is a best practice for large models and multi-models:\n",
    "     The data model defines the tuples and tuple sets that will contain the input and output data.\n",
    "     The optimization model defines the decision variables, decision expressions, objective function, \n",
    "     constraints, and pre- and post-processing data transformations.\n",
    "     \n",
    "     When the OPL model consists of multiple components, ModelSource concatenates them in the order\n",
    "     presented, and it is not necessary to use OPL include statements to import the components.\n",
    "     The multiple model files need not be located in the same resource folder.\n",
    "     \n",
    "     Note: developers generally need not use this class directly. Instead, it is recommended\n",
    "     to use the setOPLModel method of the Optimizer class.\n",
    "    '''\n",
    "    \n",
    "    def __init__(self, name= \"OPL.mod\", dotMods= None, modelText= None):\n",
    "        '''\n",
    "         Creates a new ModelSource instance from URLs pointing to OPL .mod files.\n",
    "         This method can take any number of URL arguments, but\n",
    "         there are two common use cases:\n",
    "         First, the optimization model can be composed of two pieces: \n",
    "         A data model that defines the tuples and tuple sets that will contain the input and output data.\n",
    "         An optimization model that defines the decision variables, decision expressions, objective function, \n",
    "         constraints, and pre- and post-processing data transformations.\n",
    "         The two are concatenated, so they must be presented in that order.\n",
    "         If such a composite model is used, you do not need to import the data model into the optimization model using an OPL include statement.\n",
    "         Second, you do not have to use a separate data model, in which case a single model URL must be provided \n",
    "         which encompasses both the data model and the optimization model.\n",
    "        \n",
    "        @param name: the name assigned to this OPL model (should have the format of a file name with a .mod extension)\n",
    "        @type name: String\n",
    "        @param dotMods: URLs pointing to OPL .mod files, which will be concatenated in the order given\n",
    "        @type dotMods: List<URL>\n",
    "        @param modelText: the text of the OPL model, which will be concatenated in the order given\n",
    "        @type modelText: List<String>\n",
    "        @raise: ValueError if dotMods or modelText is empty\n",
    "        '''\n",
    "        \n",
    "        self.name= name;\n",
    "        if dotMods is not None and not dotMods: #is empty\n",
    "            raise ValueError(\"argument cannot be empty\");\n",
    "        self.dotMods= dotMods;\n",
    "        if modelText is not None and not modelText: #is empty\n",
    "            raise ValueError(\"argument cannot be empty\");\n",
    "        self.modelText= modelText;\n",
    "        \n",
    "    def getName(self):\n",
    "        '''\n",
    "         @return:  the name assigned to this OPL model\n",
    "         @type String \n",
    "        '''\n",
    "        return self.name\n",
    "    \n",
    "    def isEmpty(self):\n",
    "        '''\n",
    "         @return true if both dotMods and modelText are null; false otherwise\n",
    "        '''\n",
    "        return self.dotMods is None and self.modelText is None\n",
    "    \n",
    "    def toStream(self):\n",
    "        '''\n",
    "         Concatenates the model components and creates an input file for reading them.\n",
    "         \n",
    "         @return a file\n",
    "        '''\n",
    "        if self.dotMods: #is not empty        \n",
    "            result= fileinput.input((urllib.FancyURLopener(f) for f in self.dotMods))\n",
    "            return result           \n",
    "        if self.modelText: #is not empty\n",
    "            result= cStringIO.StringIO(\"\".join(self.modelText)) \n",
    "            return result           \n",
    "        raise ValueError(\"model source is empty\")\n",
    " \n",
    "# end class ModelSource"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.7.3 Setting Up and Submitting the Solve Job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the Optimizer class, the following solves the Warehousing problem:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "u'OPTIMAL_SOLUTION'"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "problem= Optimizer(\"Warehousing\", credentials={\"url\":url, \"key\":key})\\\n",
    "        .setOPLModel(\"Warehousing.mod\", modelText= [warehousing_data_dotmod, warehousing_inputs, warehousing_dotmod, warehousing_outputs])\\\n",
    "        .setResultDataModel(warehousingResultDataModel)\n",
    "warehousingResult= problem.solve(warehousingData.copy(\"warehousingDataNoCoord\", \"warehouses\", \"routes\", \"stores\", \"demands\", \"scenarios\"))\n",
    "# Note: the mapCoordinates table is not used in the optimization and so is not sent to the optimizer\n",
    "problem.getSolveStatus()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.7.4 Retrieving the Optimal Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "collector: WarehousingResult\n",
      "table: objectives\n",
      "+-----------+-------------+----------+---------+--------------------+\n",
      "|problem    |dExpr        |scenarioId|iteration|value               |\n",
      "+-----------+-------------+----------+---------+--------------------+\n",
      "|Warehousing|capitalCost  |Nominal   |0        |6373620.0           |\n",
      "|Warehousing|operatingCost|Nominal   |0        |4580688.489999998   |\n",
      "|Warehousing|totalCost    |Nominal   |0        |1.0954308489999998E7|\n",
      "+-----------+-------------+----------+---------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "warehousingResult.displayTable(\"objectives\", sys.stdout);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "collector: WarehousingResult\n",
      "table: openWarehouses\n",
      "+-----------------+----------+---------+----+--------+\n",
      "|         location|scenarioId|iteration|open|capacity|\n",
      "+-----------------+----------+---------+----+--------+\n",
      "|     New York, NY|   Nominal|        0|   1|  3961.0|\n",
      "|Lawrenceville, GA|   Nominal|        0|   1|  1146.0|\n",
      "|      Chicago, IL|   Nominal|        0|   1|  2720.0|\n",
      "|       Dallas, TX|   Nominal|        0|   1|  1395.0|\n",
      "|       Denver, CO|   Nominal|        0|   1|   874.0|\n",
      "|  Los Angeles, CA|   Nominal|        0|   1|  5581.0|\n",
      "|San Francisco, CA|   Nominal|        0|   1|  2388.0|\n",
      "+-----------------+----------+---------+----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "openWarehouses= warehousingResult.getTable(\"openWarehouses\").select('*').where(\"open == 1\")\n",
    "print(\"collector: WarehousingResult\")\n",
    "print(\"table: openWarehouses\")\n",
    "openWarehouses.show(openWarehouses.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3 Transforming an Optimization Problem to Tableau Form"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The simplex tableau form is almost the problem format used by the solver. The modeling layer automatically transforms an optimization problem such as the Warehousing example to tableau form and recovers the solution, transparently to the model developer. This section discusses how these transformations occur, using the Warehousing problem as an example."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1 Using Relational Database Operations to Reshape the Instance Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Many readers will have noticed at this point the correspondence between the data structures of OPL and those of the relational model commonly used in database programming. A tuple set in OPL corresponds to a table in a relational database. A tuple in OPL corresponds to a row, or record, in a database table. A data set, or *collector*, for an optimization model consists of one or more tables (tuple sets), usually having foreign keys that link them together. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This correspondence is more than casual; it reflects a deep relationship between optimization modeling and the structure of data. This relationship, in turn, implies that the common operations on relational data, namely *selects* and *joins*, can also be used to transform the data used in optimization modeling. Since relational database systems are designed to execute these operations very efficiently on very large datasets, using such a system as the basis for optimization modeling can have very great advantages in speed and efficiency. Not all of the latency in solving an optimization problem is due to the computational algorithms; many times, the upstream and downstream data transformations contribute significantly as well."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This section illustrates that insight. It builds the transformations described in the preceding section using a database system, in this case Apache Spark (although other database systems could be used just as well). See http://spark.apache.org/. Spark was chosen because it is designed for use with very large, distributed datasets, called *Resilient Distributed Datasets (RDD)*. While an RDD does not have to have a relational structure, for the purposes of this paper, relational structure will be assumed. A Spark RDD with a relational structure is called a *dataset* (or *dataframe* in Python)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Identify the input dataframes\n",
    "warehouses = warehousingData.getTable('warehouses')\n",
    "stores =     warehousingData.getTable('stores')\n",
    "routes =     warehousingData.getTable('routes')\n",
    "demands =    warehousingData.getTable('demands')\n",
    "scenarios =  warehousingData.getTable('scenarios')\n",
    "#scenarios is a singleton (has a single row)\n",
    "scenarioId = scenarios.first()[\"id\"]\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.1 The Tableau Data Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The tableau data model is reprinted below for reference."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      " //Defines a column for a boolean variable\n",
      " tuple BooleanColumn {\n",
      "    key string variable;    //name = variable+index\n",
      "    int lower;              //lower bound (always 0)\n",
      "    int upper;              //upper bound (always 1)\n",
      "    int value;              //optimal value (output only)\n",
      " }\n",
      " \n",
      " //Defines a column for an integer variable\n",
      " tuple IntegerColumn {\n",
      "    key string variable;    //name = variable+index\n",
      "    int lower;              //lower bound\n",
      "    int upper;              //upper bound\n",
      "    int value;              //optimal value (output only)\n",
      " }\n",
      " \n",
      " //Defines a column for a continuous variable\n",
      " tuple FloatColumn {\n",
      "    key string variable;    //name = variable+index\n",
      "    float lower;            //lower bound\n",
      "    float upper;            //upper bound\n",
      "    float value;            //optimal value (output only)\n",
      " }\n",
      " \n",
      " //Defines a row for a constraint or a decision expression\n",
      " tuple Row {\n",
      "    key string cnstraint;   //name = constraint+index  \n",
      "    string sense;           //GE(>=), EQ(==), LE(<=) or dexpr (must be dexpr for a decision expression)\n",
      "    float rhs;              //right-hand side term (must be zero for a decision expression)\n",
      " }\n",
      "  \n",
      " //Defines a coefficient at a specific row and column\n",
      " tuple Entry {\n",
      "    key string cnstraint;   //name = constraint+index   (can also be used for a decision expression) \n",
      "    key string variable;    //name = variable+index  \n",
      "    float coefficient;      //coefficent\n",
      " }\n",
      " \n",
      " //Defines a decision expression value\n",
      " tuple Objective {\n",
      "    key string name;        //must correspond to the name of one of the decision expressions\n",
      "    string sense;           //minimize or maximize\n",
      "    float value;            //optimal decision expression value (output only)\n",
      " }\n",
      " \n"
     ]
    }
   ],
   "source": [
    "print(tableau_data_model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The corresponding Spark schemas are built with an <code>OPLCollector</code> instance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Create the tableau data model\n",
    "tableauData= OPLCollector(\"tableauData\")\n",
    "tableauADMBuilder= tableauData.buildADM()\n",
    "tableauADMBuilder.addSchema(\"integerColumns\", buildSchema(\n",
    "    (\"variable\", StringType()),\n",
    "    (\"lower\", IntegerType()),\n",
    "    (\"upper\", IntegerType()),\n",
    "    (\"value\", IntegerType())))\n",
    "tableauADMBuilder.addSchema(\"booleanColumns\", SchemaBuilder()\\\n",
    "    .copyFields(tableauADMBuilder.referenceSchema(\"integerColumns\"))\\\n",
    "    .buildSchema())\n",
    "tableauADMBuilder.addSchema(\"floatColumns\", buildSchema(\n",
    "    (\"variable\", StringType()),\n",
    "    (\"lower\", DoubleType()),\n",
    "    (\"upper\", DoubleType()),\n",
    "    (\"value\", DoubleType())))\n",
    "tableauADMBuilder.addSchema(\"rows\", buildSchema(\n",
    "    (\"cnstraint\", StringType()),\n",
    "    (\"sense\", StringType()),\n",
    "    (\"rhs\", DoubleType())))\n",
    "tableauADMBuilder.addSchema(\"entries\", buildSchema(\n",
    "    (\"cnstraint\", StringType()),\n",
    "    (\"variable\", StringType()),\n",
    "    (\"coefficient\", DoubleType())))\n",
    "tableauADMBuilder.addSchema(\"objectives\", buildSchema(\n",
    "    (\"name\", StringType()),\n",
    "    (\"sense\", StringType()),\n",
    "    (\"value\", DoubleType())))\n",
    "tableauDataModel= tableauADMBuilder.build()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.2 The Transformation Data Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is necessary to transform the instance data for the warehousing model into the tableau data model. This transformation occurs in several steps which are mediated by a transitional collector. The schema of this collector is specified in the next cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the data model to transform the warehousing data into the tableau\n",
    "tableauTransformations = OPLCollector(\"tableauTransformations\")\n",
    "tableauTransformationsADMBuilder = tableauTransformations.buildADM()\n",
    "tableauTransformationsADMBuilder.addSchema(\"columns_open\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"booleanColumns\"))\\\n",
    "    .addField(\"location\", StringType())\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"columns_capacity\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"floatColumns\"))\\\n",
    "    .addField(\"location\", StringType())\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"columns_ship\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"floatColumns\")) \\\n",
    "    .addField(\"location\", StringType())\\\n",
    "    .addField(\"store\", StringType())\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"rows_ctCapacity\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"rows\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"rows_ctDemand\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"rows\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"rows_ctSupply\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"rows\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"rows_dexpr\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"rows\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_ctCapacity_capacity\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_ctCapacity_ship\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_ctDemand_ship\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_ctSupply_open\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_ctSupply_ship\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_dexpr_open\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_dexpr_capacity\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsADMBuilder.addSchema(\"entries_dexpr_ship\", SchemaBuilder()\\\n",
    "    .copyFields(tableauData.getADM().get(\"entries\"))\\\n",
    "    .buildSchema())\n",
    "tableauTransformationsDataModel= tableauTransformationsADMBuilder.build()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.2 Encoding the Decision Variables"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first step is to map the decision variables to columns in the tableau. The purpose of this encoding is two-fold. First, it associates the data in the application data model with the schema of the tableau data model. Second it establishes a mapping between the keys in the application data model (which are specifc to its domain) and the column indices (i.e. the keys) of the tableau, which need to be independent of the application data schema. This mapping enables two-way communication between the application layer and the solving layer.\n",
    "\n",
    "The Warehousing example has three sets of decision variables: \n",
    "<ul>\n",
    "    <li><code>dvar boolean open[warehouses]</code></li> \n",
    "    <li><code>dvar float capacity[warehouses] in 0..infinity</code></li> \n",
    "    <li><code>dvar float ship[routes] in 0.0..1.0</code></li>\n",
    "</ul>\n",
    "\n",
    "The column indicies are simply computed as the name of the decision variable appended with the key in the underlying index table, e.g. <code>\"open_Brockton, MA\"</code>."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<__main__.DataBuilder at 0x7f5db5b202d0>"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tableauTransformer = tableauTransformations.buildData()\n",
    "\n",
    "# Encode the columns\n",
    "tableauTransformer.addTable(\"columns_open\",\n",
    "    warehouses.select(\"location\")\\\n",
    "        .withColumn(\"variable\", functions.concat(functions.lit(\"open_\"), warehouses[\"location\"]))\\\n",
    "        .withColumn(\"upper\", functions.lit(1))\\\n",
    "        .withColumn(\"lower\", functions.lit(0))\\\n",
    "        .withColumn(\"value\", functions.lit(0)))\n",
    "tableauTransformer.addTable(\"columns_capacity\",\n",
    "    warehouses.select(\"location\")\\\n",
    "        .withColumn(\"variable\", functions.concat(functions.lit(\"capacity_\"), warehouses[\"location\"]))\\\n",
    "        .withColumn(\"upper\", functions.lit(1.0e20))\\\n",
    "        .withColumn(\"lower\", functions.lit(0.0))\\\n",
    "        .withColumn(\"value\", functions.lit(0.0)))\n",
    "tableauTransformer.addTable(\"columns_ship\",\n",
    "    routes.select(\"location\", \"store\")\\\n",
    "        .withColumn(\"variable\", functions.concat(functions.lit(\"ship_\"), routes[\"location\"], functions.lit(\"_\"),\n",
    "                                                 routes[\"store\"]))\\\n",
    "        .withColumn(\"upper\", functions.lit(1.0))\\\n",
    "        .withColumn(\"lower\", functions.lit(0.0))\\\n",
    "        .withColumn(\"value\", functions.lit(0.0)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.3 Encoding the Constraints and Decision Expressions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The next  step is to map the constraints to rows in the tableau. An encoding similar to that used for the decision variables applies to the three sets of constraints: \n",
    "<ul>\n",
    "    <li><code>constraint ctCapacity[warehouses]</code></li> \n",
    "    <li><code>constraint ctDemand[stores]</code></li> \n",
    "    <li><code>constraint ctSupply[routes]</code></li>\n",
    "</ul>\n",
    "\n",
    "Also encoded are the rows corresponding to the decision expressions:\n",
    "<ul>\n",
    "    <li><code>dexpr float capitalCost</code></li>\n",
    "    <li><code>dexpr float operatingCost</code></li>\n",
    "    <li><code>dexpr float totalCost</code></li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<__main__.DataBuilder at 0x7f5db5b202d0>"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tableauTransformer.addTable(\"rows_ctCapacity\",\n",
    "    warehouses.select(\"location\")\\\n",
    "        .withColumn(\"cnstraint\", functions.concat(functions.lit(\"ctCapacity_\"), warehouses[\"location\"]))\\\n",
    "        .withColumn(\"sense\", functions.lit(\"GE\"))\\\n",
    "        .withColumn(\"rhs\", functions.lit(0.0)))\n",
    "tableauTransformer.addTable(\"rows_ctDemand\",\n",
    "    stores.select(\"storeId\")\\\n",
    "        .withColumn(\"cnstraint\", functions.concat(functions.lit(\"ctDemand_\"), stores[\"storeId\"]))\\\n",
    "        .withColumn(\"sense\", functions.lit(\"GE\"))\\\n",
    "        .withColumn(\"rhs\", functions.lit(1.0))\\\n",
    "        .withColumnRenamed(\"storeId\", \"store\"))\n",
    "tableauTransformer.addTable(\"rows_ctSupply\",\n",
    "    routes.select(\"location\", \"store\")\\\n",
    "        .withColumn(\"cnstraint\", functions.concat(functions.lit(\"ctSupply_\"), routes[\"location\"], functions.lit(\"_\"),\n",
    "                                                  routes[\"store\"]))\\\n",
    "        .withColumn(\"sense\", functions.lit(\"GE\"))\\\n",
    "        .withColumn(\"rhs\", functions.lit(0.0)))\n",
    "tableauTransformer.addTable(\"rows_dexpr\",\n",
    "    SPARK_SESSION.createDataFrame(\n",
    "        [   Row(cnstraint= \"capitalCost\", sense= \"dexpr\", rhs= 0.0),\n",
    "            Row(cnstraint= \"operatingCost\", sense= \"dexpr\", rhs= 0.0),\n",
    "            Row(cnstraint= \"totalCost\", sense= \"dexpr\", rhs= 0.0)],\n",
    "        tableauTransformations.getADM().get(\"rows_dexpr\"))\\\n",
    "    .select(\"cnstraint\", \"sense\", \"rhs\"))    #orders the columns properly"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.4 Reshaping the Coefficient Data into the Tableau"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The next step is to populate the coefficient matrix of the tableau with the data from the problem instance, called *reshaping* for reasons that will become clear. The table below schematically represents the coefficient matrix. The row and column labels (the left-most column and the top row) have already been defined in the preceding sections. The sparsity of the matrix is already apparent in the diagram, as three of the cells are empty. Furthermore, the sub-matrices in the non-empty cells are also themselves sparse. The construction of the tableau procedes for each sub-matrix individually."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "figure2= \"\"\"\n",
    "+-----------------+-----------------------+-----------------------------+-------------------------+\n",
    "|                 | columns_open          | columns_capacity            | columns_ship            |\n",
    "+=================+=======================+=============================+=========================+\n",
    "| rows_dexpr      | entries_dexpr_open    | entries_dexpr_capacity      | entries_dexpr_ship      |\n",
    "+-----------------+-----------------------+-----------------------------+-------------------------+\n",
    "| rows_ctCapacity |                       | entries_ctCapacity_capacity | entries_ctCapacity_ship |\n",
    "+-----------------+-----------------------+-----------------------------+-------------------------+\n",
    "| rows_ctDemand   |                       |                             | entries_ctDemand_ship   |\n",
    "+-----------------+-----------------------+-----------------------------+-------------------------+\n",
    "| rows_ctSupply   | entries_ctSupply_open |                             | entries_ctSupply_ship   |\n",
    "+-----------------+-----------------------+-----------------------------+-------------------------+\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# @hidden_cell\n",
    "# Generates the table shown above\n",
    "from tabulate import tabulate\n",
    "\n",
    "tableauColumns= ['', 'columns_open', 'columns_capacity', 'columns_ship']\n",
    "tableauDiagram= [['rows_dexpr', 'entries_dexpr_open','entries_dexpr_capacity', 'entries_dexpr_ship'],\n",
    "                 ['rows_ctCapacity', '','entries_ctCapacity_capacity', 'entries_ctCapacity_ship'],\n",
    "                 ['rows_ctDemand', '', '', 'entries_ctDemand_ship'],\n",
    "                 ['rows_ctSupply', 'entries_ctSupply_open', '', 'entries_ctSupply_ship']]\n",
    "\n",
    "\n",
    "#print tabulate(tableauDiagram, tableauColumns, tablefmt=\"grid\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These transformations associate the row and column entries in the tableau with the coefficients of the constraint inequalities. Recall that the constraints of the Warehousing model are:\n",
    "<p>\n",
    "<code>\n",
    " dexpr float capitalCost=   sum(w in warehouses) (w.fixedCost X open[w] + w.capacityCost X capacity[w]);\n",
    " dexpr float operatingCost= sum(r in routes) r.shippingCost X demand[r] X ship[r];\n",
    " dexpr float totalCost=     sum(w in warehouses) (w.fixedCost X open[w] + w.capacityCost X capacity[w]) +\n",
    "                            sum(r in routes) r.shippingCostXdemand[r] X ship[r];\n",
    "forall(w in warehouses)\n",
    "// Cannot ship more out of a warehouse than its capacity\n",
    "   ctCapacity[w]: capacity[w] - sum(r in routes: r.location==w.location) demand[r] X ship[r] >= 0.0;\t\n",
    "forall(s in stores)\n",
    "// Must ship at least 100% of each store's demand\n",
    "   ctDemand[s]: sum(r in routes: r.store==s.storeId) ship[r] >= 1.0;  \n",
    "forall(r in routes, w in warehouses: w.location==r.location)\n",
    "// Can only ship along a supply route if its warehouse is open\t  \n",
    "   ctSupply[r]: open[w] - ship[r] >= 0.0;\n",
    "</code>\n",
    "<p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The SQL statements (using Spark SQL) corresponding to the OPL are as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'columns_capacity': DataFrame[location: string, variable: string, upper: double, lower: double, value: double],\n",
       " 'columns_open': DataFrame[location: string, variable: string, upper: int, lower: int, value: int],\n",
       " 'columns_ship': DataFrame[location: string, store: string, variable: string, upper: double, lower: double, value: double],\n",
       " 'entries_ctCapacity_capacity': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_ctCapacity_ship': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_ctDemand_ship': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_ctSupply_open': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_ctSupply_ship': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_dexpr_capacity': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_dexpr_open': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'entries_dexpr_ship': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'rows_ctCapacity': DataFrame[location: string, cnstraint: string, sense: string, rhs: double],\n",
       " 'rows_ctDemand': DataFrame[store: string, cnstraint: string, sense: string, rhs: double],\n",
       " 'rows_ctSupply': DataFrame[location: string, store: string, cnstraint: string, sense: string, rhs: double],\n",
       " 'rows_dexpr': DataFrame[cnstraint: string, sense: string, rhs: double]}"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tableauTransformer.addTable(\n",
    "    \"entries_ctCapacity_capacity\",\n",
    "    tableauTransformer.referenceTable(\"rows_ctCapacity\")\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_capacity\"), on=\"location\")\\\n",
    "        .select(\"cnstraint\", \"variable\")\\\n",
    "        .withColumn(\"coefficient\", functions.lit(1.0)))\n",
    "# demand at the store at the end of each route\n",
    "demandOnRoute = routes\\\n",
    "        .join(demands.where(demands[\"scenarioId\"] == functions.lit(scenarioId)), on=\"store\")\\\n",
    "        .select(\"location\", \"store\", \"amount\")\\\n",
    "        .withColumnRenamed(\"amount\", \"demand\")\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_ctCapacity_ship\",\n",
    "    tableauTransformer.referenceTable(\"rows_ctCapacity\")\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_ship\"), on=\"location\")\\\n",
    "        .join(demandOnRoute, on=[\"location\", \"store\"])\\\n",
    "        .withColumn(\"coefficient\", -demandOnRoute[\"demand\"])\\\n",
    "        .select(\"cnstraint\", \"variable\", \"coefficient\"))\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_ctDemand_ship\",\n",
    "    tableauTransformer.referenceTable(\"rows_ctDemand\")\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_ship\"), on=\"store\")\\\n",
    "        .select(\"cnstraint\", \"variable\")\\\n",
    "        .withColumn(\"coefficient\", functions.lit(1.0)))\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_ctSupply_open\",\n",
    "    tableauTransformer.referenceTable(\"rows_ctSupply\")\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_open\"), on=\"location\")\\\n",
    "        .select(\"cnstraint\", \"variable\")\\\n",
    "        .withColumn(\"coefficient\", functions.lit(1.0)))\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_ctSupply_ship\",\n",
    "    tableauTransformer.referenceTable(\"rows_ctSupply\")\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_ship\"), on=[\"location\", \"store\"])\\\n",
    "        .select(\"cnstraint\", \"variable\")\\\n",
    "        .withColumn(\"coefficient\", functions.lit(-1.0)))\n",
    "rows_dexpr = tableauTransformer.referenceTable(\"rows_dexpr\")\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_dexpr_open\",\n",
    "    (rows_dexpr.where((rows_dexpr[\"cnstraint\"] == functions.lit(\"capitalCost\"))\\\n",
    "                    | (rows_dexpr[\"cnstraint\"] == functions.lit(\"totalCost\"))))\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_open\").join(warehouses, on=\"location\"), \n",
    "              how=\"cross\")\\\n",
    "        .select(\"cnstraint\", \"variable\", \"fixedCost\")\\\n",
    "        .withColumnRenamed(\"fixedCost\", \"coefficient\"))\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_dexpr_capacity\",\n",
    "    (rows_dexpr.where((rows_dexpr[\"cnstraint\"] == functions.lit(\"capitalCost\"))\\\n",
    "                    | (rows_dexpr[\"cnstraint\"] == functions.lit(\"totalCost\"))))\\\n",
    "        .join(tableauTransformer.referenceTable(\"columns_capacity\").join(warehouses, on=\"location\"), \n",
    "              how=\"cross\")\\\n",
    "        .select(\"cnstraint\", \"variable\", \"capacityCost\")\\\n",
    "        .withColumnRenamed(\"capacityCost\", \"coefficient\"))\n",
    "tableauTransformer.addTable(\n",
    "    \"entries_dexpr_ship\",\n",
    "    (rows_dexpr.where((rows_dexpr[\"cnstraint\"] == functions.lit(\"operatingCost\"))\\\n",
    "                    | (rows_dexpr[\"cnstraint\"] == functions.lit(\"totalCost\"))))\\\n",
    "        .join((tableauTransformer.referenceTable(\"columns_ship\")\\\n",
    "                .join((routes.join(demandOnRoute, on=[\"location\", \"store\"])\\\n",
    "                      .withColumn(\"coefficient\", demandOnRoute[\"demand\"] * routes[\"shippingCost\"])),\n",
    "                      on=[\"location\", \"store\"])), \n",
    "              how=\"cross\")\\\n",
    "        .select(\"cnstraint\", \"variable\", \"coefficient\"))\n",
    "\n",
    "tableauTransformer.build()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.5 Creating the Tableau Input Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The input tables are inserted into an <code>OPLCollector</code> instance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'booleanColumns': DataFrame[variable: string, upper: int, lower: int, value: int],\n",
       " 'entries': DataFrame[cnstraint: string, variable: string, coefficient: double],\n",
       " 'floatColumns': DataFrame[variable: string, upper: double, lower: double, value: double],\n",
       " 'integerColumns': DataFrame[variable: string, lower: int, upper: int, value: int],\n",
       " 'objectives': DataFrame[name: string, sense: string, value: double],\n",
       " 'rows': DataFrame[cnstraint: string, sense: string, rhs: double]}"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Drop the instance-specific keys (location and store), which are not supported in the tableau model\n",
    "tableauData.buildData()\\\n",
    "    .addTable(\"booleanColumns\",\n",
    "           tableauTransformations.getTable(\"columns_open\").drop(\"location\"))\\\n",
    "    .addTable(\"floatColumns\",\n",
    "               tableauTransformations.getTable(\"columns_capacity\").drop(\"location\")\\\n",
    "        .union(tableauTransformations.getTable(\"columns_ship\").drop(\"location\").drop(\"store\")))\\\n",
    "    .addEmptyTable(\"integerColumns\")\\\n",
    "    .addTable(\"rows\",\n",
    "               tableauTransformations.getTable(\"rows_ctCapacity\").drop(\"location\")\\\n",
    "        .union(tableauTransformations.getTable(\"rows_ctDemand\").drop(\"store\"))\\\n",
    "        .union(tableauTransformations.getTable(\"rows_ctSupply\").drop(\"location\").drop(\"store\"))\\\n",
    "        .union(tableauTransformations.getTable(\"rows_dexpr\")))\\\n",
    "    .addTable(\"entries\",\n",
    "               tableauTransformations.getTable(\"entries_ctSupply_open\")\\\n",
    "        .union(tableauTransformations.getTable(\"entries_ctSupply_ship\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_ctCapacity_capacity\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_ctCapacity_ship\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_ctDemand_ship\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_dexpr_open\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_dexpr_capacity\"))\\\n",
    "        .union(tableauTransformations.getTable(\"entries_dexpr_ship\")))\\\n",
    "    .addTable(\"objectives\",\n",
    "        SPARK_SESSION.createDataFrame(\n",
    "            [Row(name= \"totalCost\", sense= \"minimize\", value= 0.0)],\n",
    "            tableauData.getADM().get(\"objectives\"))\n",
    "        .select(\"name\", \"sense\", \"value\"))\\\n",
    ".build()\n",
    "# note: the select clause in objectives table is needed to insure the order of the columns so that the JSON serialization works properly"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.6 Solving the Tableau Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once these transformations have been applied, the tableau form of the model shown in section 1, <code>tableau_optimization_problem</code>, can be solved:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "u'OPTIMAL_SOLUTION'"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tableauProblem = Optimizer(\"TableauProblem\", credentials={\"url\": url, \"key\": key})\\\n",
    "    .setOPLModel(\"TableauProblem.mod\",\n",
    "                 modelText=[tableau_data_model, tableau_inputs, tableau_optimization_problem, tableau_outputs])\\\n",
    "    .setResultDataModel(ADMBuilder()\\\n",
    "        .addSchema(\"booleanDecisions\", tableauData.getSchema(\"booleanColumns\"))\\\n",
    "        .addSchema(\"integerDecisions\", tableauData.getSchema(\"integerColumns\"))\\\n",
    "        .addSchema(\"floatDecisions\", tableauData.getSchema(\"floatColumns\"))\\\n",
    "        .addSchema(\"optimalObjectives\", tableauData.getSchema(\"objectives\"))\\\n",
    "        .build())\n",
    "tableauResult = tableauProblem.solve(tableauData)\n",
    "tableauProblem.getSolveStatus()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, it must be emphasized that the tableau model, whether realized in OPL or Python (or any other language for that matter) is independent of the structure of the model instance represented by the modeling layer. The modeling layer constructs the tableau data structures through the series of SQL transformations discussed above. Therefore, ordinarily, the model developer need not concern herself with the tableau model, which is not altered during the development process."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.7 Recovering the Warehousing Solution"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the solver has run, the solution must be mapped back from the tableau form to the original decision variables. As discussed in section 1, this step also entails rehaping the solution arrays into tuple sets. The following transformations are used:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "collector: warehousingResult\n",
      "table: objectives\n",
      "+-------------+-----------------+-----------+----------+---------+\n",
      "|dExpr        |value            |problem    |scenarioId|iteration|\n",
      "+-------------+-----------------+-----------+----------+---------+\n",
      "|capitalCost  |6373620.0        |warehousing|Nominal   |0        |\n",
      "|operatingCost|4580688.489999998|warehousing|Nominal   |0        |\n",
      "|totalCost    |1.095430849E7    |warehousing|Nominal   |0        |\n",
      "+-------------+-----------------+-----------+----------+---------+\n",
      "\n",
      "collector: warehousingResult\n",
      "table: openWarehouses\n",
      "+-----------------+----+--------+----------+---------+\n",
      "|location         |open|capacity|scenarioId|iteration|\n",
      "+-----------------+----+--------+----------+---------+\n",
      "|San Francisco, CA|1   |2388.0  |Nominal   |0        |\n",
      "|Los Angeles, CA  |1   |5581.0  |Nominal   |0        |\n",
      "|Dallas, TX       |1   |1395.0  |Nominal   |0        |\n",
      "|Chicago, IL      |1   |2720.0  |Nominal   |0        |\n",
      "|New York, NY     |1   |3961.0  |Nominal   |0        |\n",
      "|Lawrenceville, GA|1   |1146.0  |Nominal   |0        |\n",
      "|Denver, CO       |1   |874.0   |Nominal   |0        |\n",
      "+-----------------+----+--------+----------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "warehousingResult = OPLCollector(\"warehousingResult\", warehousingResultDataModel)\n",
    "resultsBuilder = warehousingResult.buildData()\n",
    "resultsBuilder.addTable(\"objectives\",\n",
    "    tableauResult.getTable(\"optimalObjectives\").select(\"name\", \"value\")\\\n",
    "    .withColumnRenamed(\"name\", \"dExpr\")\\\n",
    "    .withColumn(\"problem\", functions.lit(\"warehousing\"))\\\n",
    "    .withColumn(\"scenarioId\", functions.lit(scenarioId))\\\n",
    "    .withColumn(\"iteration\", functions.lit(0)))\n",
    "resultsBuilder.addTable(\"openWarehouses\",\n",
    "    (tableauResult.getTable(\"booleanDecisions\").select(\"variable\", \"value\").withColumnRenamed(\"value\", \"open\")\\\n",
    "    .join(tableauTransformations.getTable(\"columns_open\"), on=\"variable\")).drop(\"variable\")\\\n",
    "    .join(\n",
    "        tableauResult.getTable(\"floatDecisions\").select(\"variable\", \"value\").withColumnRenamed(\"value\", \"capacity\")\\\n",
    "            .join(tableauTransformations.getTable(\"columns_capacity\"), on=\"variable\").drop(\"variable\"),\n",
    "        on=\"location\")\n",
    "    .select(\"location\", \"open\", \"capacity\")\\\n",
    "    .where(\"open > 0\")\\\n",
    "    .withColumn(\"scenarioId\", functions.lit(scenarioId))\\\n",
    "    .withColumn(\"iteration\", functions.lit(0)))\n",
    "floatDecisions = tableauResult.getTable(\"floatDecisions\").select(\"variable\", \"value\")\n",
    "resultsBuilder.addTable(\"shipments\",\n",
    "    floatDecisions\\\n",
    "    .join(tableauTransformations.getTable(\"columns_ship\"), on=\"variable\").drop(\"variable\")\\\n",
    "    .join(demandOnRoute, on=[\"location\", \"store\"])\\\n",
    "    .withColumn(\"amount\", demandOnRoute[\"demand\"]*(floatDecisions[\"value\"]))\\\n",
    "    .select(\"location\", \"store\", \"amount\")\\\n",
    "    .where(\"amount > 0.0\")\\\n",
    "    .withColumn(\"scenarioId\", functions.lit(scenarioId))\\\n",
    "    .withColumn(\"iteration\", functions.lit(0)))\n",
    "resultsBuilder.build()\n",
    "\n",
    "warehousingResult.displayTable(\"objectives\")\n",
    "warehousingResult.displayTable(\"openWarehouses\")\n",
    "#to see the lengthy shipments table, uncomment the next line\n",
    "#warehousingResult.displayTable(\"shipments\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As can readily be seen, this solution is identical to the obtained by directly solving the warehousing model above."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4 Conclusion: Equivalence of Tuple Slicing and SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In OPL, operations such as this constraint specification\n",
    "<code>\n",
    "\tforall(s in stores)\n",
    "\t  ctDemand[s]: sum(r in routes: r.store==s.storeId) ship[r] >= 1.0;\n",
    "</code>\n",
    "are called *tuple slicing*, in which a filtering condition is applied within an iteration over index sets.\n",
    "\n",
    "Section 3.4 implemented such slicing operations using their equivalents in SQL (we have stylized the code somewhat to make the connection clearer):\n",
    "<code>\n",
    "entries_ctDemand_ship= rows_ctDemand\n",
    "        .join(columns_ship on store)\n",
    "        .select(cnstraint, variable)\n",
    "        .withColumn(coefficient= 1.0)\n",
    "</code>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A more complex example:\n",
    "<code>\n",
    "float demand[routes]= [r: d.amount | r in routes,  d in demands: r.store==d.store];\n",
    "...\n",
    "forall(w in warehouses)\n",
    " \t  ctCapacity[w]: capacity[w] >= sum(r in routes: r.location==w.location) demand[r]*ship[r];\n",
    "</code>\n",
    "is implemented as\n",
    "<code>\n",
    "demandOnRoute = \n",
    "    routes\n",
    "        .join((demands.where(demands.scenarioId == literal(scenarioId)) on store)\n",
    "        .select(location, store, amount as demand)\n",
    "entries_ctCapacity_capacity= \n",
    "   rows_ctCapacity\n",
    "        .join(columns_capacity on location)\n",
    "        .select(cnstraint, variable)\n",
    "        .withColumn(coefficient=1.0)\n",
    "entries_ctCapacity_ship=\n",
    "    rows_ctCapacity\n",
    "        .join(columns_ship on location)\n",
    "        .join(demandOnRoute on location, store)\n",
    "        .select(cnstraint, variable)\n",
    "        .withColumn(coefficient, -demandOnRoute.demand)\n",
    "</code>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "This is a general rule: every slicing operation on tuple sets used in creating an optimization model has an equivalent SQL operation on database tables (or dataframes). Thus, one can regard a modeling language such as OPL as simply a different dialect of SQL. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The primary topic addressed in this paper is to demonstrate the equivalence of data handling in an optimization modeling language to operations on relational data base tables. Additionally, it illustrates the use of Apache Spark for data handling in optimization."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Author"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Dr. Jeremy Bloom** is an offering manager for IBM Data Science Experience and an expert on decision optimization. In the course of his more than 35-year career, he has lead research programs for energy companies and developed software products using operation research to solve practical business problems. Dr. Bloom has a bachelor's degree in Electrical Engineering from Carnegie-Mellon University and a master's degree and doctorate from Massachusetts Institute of Technology in Operations Research."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## References"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " - George B. Dantzig, *Linear Programming and Extensions*, Princeton University Press (1963)\n",
    " - __[Decision Optimization on Cloud](http://dropsolve-oaas.docloud.ibmcloud.com/software/analytics/docloud)__\n",
    " - __[Decision Optimization on Cloud Python API](https://developer.ibm.com/docloud/documentation/docloud/python-api/)__\n",
    " - __[IBM Optimization Programming Language (OPL)](http://www.ibm.com/support/knowledgecenter/SSSA5P_12.7.1/ilog.odms.ide.help/OPL_Studio/maps/groupings/opl_Language.html)__"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Copyright © 2017 IBM. This notebook and its source code are released under the terms of the MIT License.**"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}