cloudarray.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# [CloudArray](https://github.com/gsd-ufal/CloudArray.jl): Easy big data programming in the cloud\n",
    "\n",
    "* GSD-UFAL Research Prototype\n",
    "* Free Software at [Github](https://github.com/gsd-ufal/CloudArray.jl)\n",
    "* Research Project with Microsoft Azure grants\n",
    "* **CloudArrayBox [Try it here!](http://cloudarraybox.cloudapp.net)**\n",
    "\n",
    "<img src=\"figures/cloudarray_overview.png\" width=\"400\"height=\"400\" />\n",
    "\n",
    "\n",
    "---\n",
    "## Usage\n",
    "\n",
    "First **download** and **load** CloudArray package:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: New definition \n",
      "    call(Type{DistributedArrays.DArray}, AbstractString, Any...) at /Users/alage/.julia/v0.4/CloudArray/src/CloudArray.jl:208\n",
      "is ambiguous with: \n",
      "    call(Type{DistributedArrays.DArray}, Any, DistributedArrays.DArray) at /Users/alage/.julia/v0.4/DistributedArrays/src/DistributedArrays.jl:91.\n",
      "To fix, define \n",
      "    call(Type{DistributedArrays.DArray}, AbstractString, DistributedArrays.DArray)\n",
      "before the new definition.\n",
      "WARNING: requiring \"CloudArray\" in module \"Main\" did not define a corresponding module.\n"
     ]
    }
   ],
   "source": [
    "#Pkg.clone(\"https://github.com/gsd-ufal/CloudArray.jl.git\")\n",
    "using CloudArray"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then configure the cloud host address and password (**cloud account credentials**):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ERROR: There is an error during SSH configuration. Please see the log for more details: cloud_setup.log\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "false"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "set_host(\"cloudarray01.cloudapp.net\",\"cloudarray@\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Main constructors\n",
    "\n",
    "\n",
    "CloudArray main constructors are very simple and can be created by using an `Array` or a file.\n",
    "\n",
    "### Creating a CloudArray from an `Array`\n",
    "\n",
    "You just need to tell `DArray` constructor which `Array` should be used to construct your CloudArray:\n",
    "\n",
    "```\n",
    "DArray(Array(...))\n",
    "```\n",
    "\n",
    "#### Example \n",
    "\n",
    "In this example, we first create the array `arr` with 100 random numbers then we create a CloudArray with the `arr` data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating container (1)...\n",
      "SSH configuration (1)... \n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning: Permanently added '[cloudarray01.cloudapp.net]:3002,[23.99.60.212]:3002' (RSA) to the list of known hosts.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Adding worker (1)...\n",
      "New worker added\n",
      "Total: 1\n",
      "elapsed time: 27.757517847 seconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: deserialization checks failed while attempting to load cache from /Users/alage/.julia/lib/v0.4/DistributedArrays.ji\n",
      "WARNING: deserialization checks failed while attempting to load cache from /Users/alage/.julia/lib/v0.4/Compat.ji\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "100-element DistributedArrays.DArray{Float64,1,Array{Float64,1}}:\n",
       " 0.179902 \n",
       " 0.874428 \n",
       " 0.741819 \n",
       " 0.783698 \n",
       " 0.850314 \n",
       " 0.101395 \n",
       " 0.286979 \n",
       " 0.586523 \n",
       " 0.880305 \n",
       " 0.354472 \n",
       " 0.0630147\n",
       " 0.21228  \n",
       " 0.741673 \n",
       " ⋮        \n",
       " 0.585604 \n",
       " 0.292259 \n",
       " 0.218412 \n",
       " 0.420352 \n",
       " 0.128016 \n",
       " 0.624189 \n",
       " 0.655954 \n",
       " 0.928349 \n",
       " 0.598007 \n",
       " 0.129592 \n",
       " 0.333042 \n",
       " 0.846903 "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "arr = rand(100)\n",
    "cloudarray_from_array = DArray(arr) # will take less than one minute"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now access any value as it would be a local array:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.76926396509018"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cloudarray_from_array[57]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating a CloudArray from a file\n",
    "\n",
    "If you are dealing with big data, i.e., your RAM memory is not enough to store your data, you can create a CloudArray from a file.\n",
    "\n",
    "```Julia\n",
    "DArray(file_path)\n",
    "```\n",
    "\n",
    "```file_path``` is the path to a text file in your local or distributed file system. All lines will be used to fill `DArray` elements sequentially. This constructor ignores empty lines.\n",
    "\n",
    "\n",
    "#### Example \n",
    " \n",
    "\n",
    "Let's first create a simple text file with 100 random numbers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "f = open(\"data.txt\",\"w+\")\n",
    "for i=1:100\n",
    "    if i==100\n",
    "        write(f,\"$(rand())\")\n",
    "    else\n",
    "        write(f,\"$(rand())\\n\")\n",
    "    end    \n",
    "end\n",
    "close(f)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then we create a CloudArray with `data.txt` file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating container (4)...\n",
      "SSH configuration (4)... \n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning: Permanently added '[cloudarray01.cloudapp.net]:3050,[23.99.60.212]:3050' (RSA) to the list of known hosts.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Adding worker (4)...\n",
      "New worker added\n",
      "Total: 2\n",
      "elapsed time: 20.620173089 seconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: deserialization checks failed while attempting to load cache from /Users/alage/.julia/lib/v0.4/Compat.ji\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "100-element DistributedArrays.DArray{Float64,1,Array{Float64,1}}:\n",
       " 0.897144  \n",
       " 0.550581  \n",
       " 0.482141  \n",
       " 0.749212  \n",
       " 0.943388  \n",
       " 0.830631  \n",
       " 0.0458245 \n",
       " 0.0235713 \n",
       " 0.537933  \n",
       " 0.16375   \n",
       " 0.531444  \n",
       " 0.269559  \n",
       " 0.953395  \n",
       " ⋮         \n",
       " 0.963455  \n",
       " 0.0886685 \n",
       " 0.13586   \n",
       " 0.00930461\n",
       " 0.880849  \n",
       " 0.553979  \n",
       " 0.327784  \n",
       " 0.688187  \n",
       " 0.952335  \n",
       " 0.532966  \n",
       " 0.726188  \n",
       " 0.883657  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cloudarray_from_file = DArray(\"data.txt\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's perform a sum operation at `cloudarray_from_file`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "52.32310720010805"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sum(cloudarray_from_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This sum was performed locally at the Master, you can exploit DArray fully parallelism with further functions such as parallel Maps (`pmap`) and Reductions. See [here more information on Parallel programming in Julia](http://docs.julialang.org/en/latest/manual/parallel-computing/). \n",
    "\n",
    "\n",
    "## Core constructor\n",
    "\n",
    "If you want to tune your CloudArray, you can directly use the CloudArray core constructor:\n",
    "\n",
    "```julia\n",
    "carray_from_task(generator::Task=task_from_text(\"test.txt\"), is_numeric::Bool=true, chunk_max_size::Int=1024*1024,debug::Bool=false)\n",
    "```\n",
    "Arguments are:\n",
    "\n",
    "* ```task_from_text``` same as ```file_path```.\n",
    "* ```is_numeric``` set to ```false``` if you need to load String instead of Float.\n",
    "* ```chunk_max_size``` sets the maximum size that is allowed for each DArray chunk.\n",
    "* ```debug``` enables debug mode.\n",
    "\n",
    "### Example\n",
    "\n",
    "As follows, we create a CloudArray by using the `data.txt` file which holds numeric values, then second argument is set to `true`. We'll set the third argument (`chunk_max_size`) to `500` so DArray chunks will not have more than 500 bytes each.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating container (1)...\n",
      "SSH configuration (1)... \n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning: Permanently added '[cloudarray01.cloudapp.net]:3053,[23.99.60.212]:3053' (RSA) to the list of known hosts.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Adding worker (1)...\n",
      "New worker added\n",
      "Total: 1\n",
      "elapsed time: 23.376325559 seconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: deserialization checks failed while attempting to load cache from /Users/alage/.julia/lib/v0.4/Compat.ji\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating container (2)...\n",
      "SSH configuration (2)... \n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Warning: Permanently added '[cloudarray01.cloudapp.net]:3054,[23.99.60.212]:3054' (RSA) to the list of known hosts.\r\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Adding worker (2)...\n",
      "New worker added\n",
      "Total: 2\n",
      "elapsed time: 22.231016267 seconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: deserialization checks failed while attempting to load cache from /Users/alage/.julia/lib/v0.4/Compat.ji\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "100-element DistributedArrays.DArray{Float64,1,Array{Float64,1}}:\n",
       " 0.112248 \n",
       " 0.894684 \n",
       " 0.80095  \n",
       " 0.0974568\n",
       " 0.102072 \n",
       " 0.272024 \n",
       " 0.960695 \n",
       " 0.406005 \n",
       " 0.960488 \n",
       " 0.801466 \n",
       " 0.927738 \n",
       " 0.0988186\n",
       " 0.415196 \n",
       " ⋮        \n",
       " 0.872471 \n",
       " 0.568069 \n",
       " 0.151638 \n",
       " 0.742604 \n",
       " 0.894044 \n",
       " 0.90042  \n",
       " 0.431675 \n",
       " 0.910004 \n",
       " 0.72825  \n",
       " 0.618541 \n",
       " 0.324643 \n",
       " 0.799405 "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "custom_cloudarray_from_file = DArray(\"data.txt\", true, 500)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's define and perform a [parallel reduction](https://www.youtube.com/watch?v=JoRn4ryMclc) at the just-created CloudArray:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n",
      "WARNING: deprecated syntax \"{a for a in b}\" at In[4]:1.\n",
      "Use \"Any[a for a in b]\" instead.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "53.682358805343725"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parallel_reduce(f,darray) = reduce(f, map(fetch, { @spawnat p reduce(f, localpart(darray)) for p in workers()} ))\n",
    "parallel_reduce(+,custom_cloudarray_from_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result is the sum of all values of `custom_cloudarray_from_file`. Each DArray chunk performed in parallel the sum of the part of the DArrau it holds. The result is sent to the Master which performs the final sum. The function `map` is used to get the values with the `fetch` function.\n",
    "\n",
    "You don't really need to know it, but if you are curious on how your data is stored, you can get further information such as:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "custom_cloudarray_from_file.chunks = RemoteRef[RemoteRef{Channel{Any}}(6,1,341),RemoteRef{Channel{Any}}(7,1,350)]"
     ]
    },
    {
     "data": {
      "text/plain": [
       "2-element Array{Int64,1}:\n",
       " 6\n",
       " 7"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "custom_cloudarray_from_file.cuts = [[1,64,101]]\n",
      "custom_cloudarray_from_file.dims = (100,)\n",
      "custom_cloudarray_from_file.indexes = [(1:63,),(64:100,)]\n",
      "custom_cloudarray_from_file.pids = [6,7]\n"
     ]
    }
   ],
   "source": [
    "@show custom_cloudarray_from_file.chunks\n",
    "@show custom_cloudarray_from_file.cuts\n",
    "@show custom_cloudarray_from_file.dims\n",
    "@show custom_cloudarray_from_file.indexes\n",
    "@show custom_cloudarray_from_file.pids"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please read [DistributedArrays documentation](https://github.com/JuliaParallel/DistributedArrays.jl) to better understand these low-level details."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Julia 0.4.0",
   "language": "julia",
   "name": "julia-0.4"
  },
  "language_info": {
   "file_extension": ".jl",
   "mimetype": "application/julia",
   "name": "julia",
   "version": "0.4.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}