<commit_before>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# This notebook is adapted from: \n",
    "https://github.com/ray-project/tutorial/tree/master/examples/sharded_parameter_server.ipynb\n",
    "\n",
    "# Sharded Parameter Servers\n",
    "\n",
    "**GOAL:** The goal of this exercise is to use actor handles to implement a sharded parameter server example for **distributed asynchronous stochastic gradient descent**.\n",
    "\n",
    "Before doing this exercise, make sure you understand the concepts from the exercise on **Actor Handles**.\n",
    "\n",
    "### Parameter Servers\n",
    "\n",
    "A parameter server is simply an object that stores the parameters (or \"weights\") of a machine learning model (this could be a neural network, a linear model, or something else). It exposes two methods: one for getting the parameters and one for updating the parameters.\n",
    "\n",
    "In a typical machine learning training application, worker processes will run in an infinite loop that does the following:\n",
    "1. Get the latest parameters from the parameter server.\n",
    "2. Compute an update to the parameters (using the current parameters and some data).\n",
    "3. Send the update to the parameter server.\n",
    "\n",
    "The workers can operate synchronously (that is, in lock step), in which case distributed training with multiple workers is algorithmically equivalent to serial training with a larger batch of data. Alternatively, workers can operate independently and apply their updates asynchronously. The main benefit of asynchronous training is that a single slow worker will not slow down the other workers. The benefit of synchronous training is that the algorithm behavior is more predictable and reproducible."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Init SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Current pyspark location is : /root/anaconda2/envs/ray_train/lib/python3.6/site-packages/pyspark/__init__.py\n",
      "Start to pack current python env\n",
      "Collecting packages...\n",
      "Packing environment at '/root/anaconda2/envs/ray_train' to '/tmp/tmp7qvxc3o2/python_env.tar.gz'\n",
      "[########################################] | 100% Completed | 34.4s\n",
      "Packing has been completed: /tmp/tmp7qvxc3o2/python_env.tar.gz\n",
      "pyspark_submit_args is:  --master yarn --deploy-mode client --archives /tmp/tmp7qvxc3o2/python_env.tar.gz#python_env --num-executors 2  --executor-cores 4 --executor-memory 2g pyspark-shell \n"
     ]
    }
   ],
   "source": [
    "from zoo.common.nncontext import init_spark_on_local, init_spark_on_yarn\n",
    "import numpy as np\n",
    "import os\n",
    "hadoop_conf_dir = os.environ.get('HADOOP_CONF_DIR')\n",
    "\n",
    "if hadoop_conf_dir:\n",
    "    sc = init_spark_on_yarn(\n",
    "    hadoop_conf=hadoop_conf_dir,\n",
    "    conda_name=os.environ.get(\"ZOO_CONDA_NAME\", \"zoo\"), # The name of the created conda-env\n",
    "    num_executor=2,\n",
    "    executor_cores=4,\n",
    "    executor_memory=\"2g\",\n",
    "    driver_memory=\"2g\",\n",
    "    driver_cores=1,\n",
    "    extra_executor_memory_for_ray=\"3g\")\n",
    "else:\n",
    "    sc = init_spark_on_local(cores = 8, conf = {\"spark.driver.memory\": \"2g\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start to launch the JVM guarding process\n",
      "JVM guarding process has been successfully launched\n",
      "Start to launch ray on cluster\n",
      "Start to launch ray on local\n",
      "Executing command: ray start --redis-address 172.16.0.158:34046 --redis-password  <PASSWORD> --num-cpus 0 --object-store-memory 400000000\n"<commit_msg>Fix init_spark_on_yarn num_executors typo (#3308)<commit_after>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# This notebook is adapted from: \n",
    "https://github.com/ray-project/tutorial/tree/master/examples/sharded_parameter_server.ipynb\n",
    "\n",
    "# Sharded Parameter Servers\n",
    "\n",
    "**GOAL:** The goal of this exercise is to use actor handles to implement a sharded parameter server example for **distributed asynchronous stochastic gradient descent**.\n",
    "\n",
    "Before doing this exercise, make sure you understand the concepts from the exercise on **Actor Handles**.\n",
    "\n",
    "### Parameter Servers\n",
    "\n",
    "A parameter server is simply an object that stores the parameters (or \"weights\") of a machine learning model (this could be a neural network, a linear model, or something else). It exposes two methods: one for getting the parameters and one for updating the parameters.\n",
    "\n",
    "In a typical machine learning training application, worker processes will run in an infinite loop that does the following:\n",
    "1. Get the latest parameters from the parameter server.\n",
    "2. Compute an update to the parameters (using the current parameters and some data).\n",
    "3. Send the update to the parameter server.\n",
    "\n",
    "The workers can operate synchronously (that is, in lock step), in which case distributed training with multiple workers is algorithmically equivalent to serial training with a larger batch of data. Alternatively, workers can operate independently and apply their updates asynchronously. The main benefit of asynchronous training is that a single slow worker will not slow down the other workers. The benefit of synchronous training is that the algorithm behavior is more predictable and reproducible."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Init SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Current pyspark location is : /root/anaconda2/envs/ray_train/lib/python3.6/site-packages/pyspark/__init__.py\n",
      "Start to pack current python env\n",
      "Collecting packages...\n",
      "Packing environment at '/root/anaconda2/envs/ray_train' to '/tmp/tmp7qvxc3o2/python_env.tar.gz'\n",
      "[########################################] | 100% Completed | 34.4s\n",
      "Packing has been completed: /tmp/tmp7qvxc3o2/python_env.tar.gz\n",
      "pyspark_submit_args is:  --master yarn --deploy-mode client --archives /tmp/tmp7qvxc3o2/python_env.tar.gz#python_env --num-executors 2  --executor-cores 4 --executor-memory 2g pyspark-shell \n"
     ]
    }
   ],
   "source": [
    "from zoo.common.nncontext import init_spark_on_local, init_spark_on_yarn\n",
    "import numpy as np\n",
    "import os\n",
    "hadoop_conf_dir = os.environ.get('HADOOP_CONF_DIR')\n",
    "\n",
    "if hadoop_conf_dir:\n",
    "    sc = init_spark_on_yarn(\n",
    "    hadoop_conf=hadoop_conf_dir,\n",
    "    conda_name=os.environ.get(\"ZOO_CONDA_NAME\", \"zoo\"), # The name of the created conda-env\n",
    "    num_executors=2,\n",
    "    executor_cores=4,\n",
    "    executor_memory=\"2g\",\n",
    "    driver_memory=\"2g\",\n",
    "    driver_cores=1,\n",
    "    extra_executor_memory_for_ray=\"3g\")\n",
    "else:\n",
    "    sc = init_spark_on_local(cores = 8, conf = {\"spark.driver.memory\": \"2g\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Start to launch the JVM guarding process\n",
      "JVM guarding process has been successfully launched\n",
      "Start to launch ray on cluster\n",
      "Start to launch ray on local\n",
      "Executing command: ray start --redis-address 172.16.0.158:34046 --redis-password  <PASSWORD> --num-cpus 0 --object-store-memory 400000000\n"
{'fraction_non_alphanumeric': 0.14811867682164373, 'fraction_numerical': 0.014095714448107309, 'mean_word_length': 3.1016317016317014, 'pattern_counts': {'":': 58, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6436480', 'n_tokens_mistral': 2931, 'n_tokens_neox': 2789, 'n_words': 831}