07-rdataframe-advanced.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "60ae38e3",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# RDataFrame advanced features\n",
    "There are many more features available with the RDataFrame that might serve your analysis needs!"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "2e8e9ace",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Working with `numpy` arrays\n",
    "- RDataFrame offers interoperability with `numpy` arrays. \n",
    "\n",
    "- It can be created from a dictionary of such arrays and it can also export its contents to the same format. \n",
    "\n",
    "- All operations are available also when using the `numpy`-based dataset.\n",
    "\n",
    "- **Note:** this support is limited to one-dimensional numpy arrays, which are directly mapped to columns in the RDataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "128e586a",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import numpy\n",
    "import ROOT\n",
    "\n",
    "np_dict = {colname: numpy.random.rand(100) for colname in [\"a\",\"b\",\"c\"]}\n",
    "\n",
    "df = ROOT.RDF.FromNumpy(np_dict)\n",
    "\n",
    "print(f\"Columns in the RDataFrame: {df.GetColumnNames()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dde693ee",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "co = df.Count()\n",
    "m_a = df.Mean(\"a\")\n",
    "\n",
    "fil1 = df.Filter(\"c < 0.7\")\n",
    "def1 = fil1.Define(\"d\", \"a+b+c\")\n",
    "h = def1.Histo1D(\"d\")\n",
    "\n",
    "c = ROOT.TCanvas()\n",
    "h.Draw()\n",
    "\n",
    "print(f\"Number of rows in the dataset: {co.GetValue()}\")\n",
    "print(f\"Average value of column a: {m_a.GetValue()}\")\n",
    "c.Draw()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "47a18efa",
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# Export the modified dataframe to a dictionary of numpy arrays\n",
    "\n",
    "np_dict_mod = def1.AsNumpy()\n",
    "\n",
    "np_dict_mod"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "b3fbd633",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Multiple concurrent RDataFrame runs\n",
    "If your analysis needs multiple RDataFrames to run (for example multiple dataset samples, data vs simulation etc.), make use of `ROOT.RDF.RunGraphs` "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f342906a",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "ROOT.EnableImplicitMT()\n",
    "treename1 = \"myDataset\"\n",
    "filename1 = \"data/collections_dataset.root\"\n",
    "treename2 = \"dataset\"\n",
    "filename2 = \"data/example_file.root\"\n",
    "\n",
    "df1 = ROOT.RDataFrame(treename1, filename1)\n",
    "df2 = ROOT.RDataFrame(treename2, filename2)\n",
    "h1 = df1.Histo1D(\"px\")\n",
    "h2 = df2.Histo1D(\"a\")\n",
    " \n",
    "\n",
    "ROOT.RDF.RunGraphs((h1, h2))\n",
    "ROOT.DisableImplicitMT()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1b715694",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "c = ROOT.TCanvas()\n",
    "h1.Draw()\n",
    "c.Draw()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "795a5420",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "c = ROOT.TCanvas()\n",
    "h2.Draw()\n",
    "c.Draw()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "799b981e",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Re-defining values of a column\n",
    "\n",
    "Sometimes you might want to keep a column with the same name, but modify its values. This is available via the `Redefine` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "374e35a3",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "df = ROOT.RDataFrame(5).Define(\"x\", \"42\")\n",
    "df.Display().Print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a7a7a553",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "df = df.Redefine(\"x\", \"x * 10\")\n",
    "df.Display().Print()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "d820208f",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Notice how we could use the previous values of the `\"x\"` column to define the new values."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "cc296f7d",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Distributed RDataFrame\n",
    "\n",
    "An `RDataFrame` analysis written in Python can be executed both *locally* - possibly in parallel on the cores of the machine - and *distributedly* by offloading computations to external resources, which include:\n",
    "\n",
    "- [Spark](https://spark.apache.org/) and \n",
    "- [Dask](https://dask.org/) clusters. \n",
    "\n",
    "- This feature is enabled by the architecture depicted below.\n",
    "\n",
    "- It shows that RDataFrame computation graphs can be mapped to different kinds of resources via backends.\n",
    "\n",
    "- In this notebook we will exercise the Dask backend, which divides an `RDataFrame` input dataset in logical ranges and submits computations for each of those ranges to Dask resources.\n",
    "\n",
    "<img src=\"images/DistRDF_architecture.png\" alt=\"Distributed RDataFrame\">"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c2731e8a",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Create a Dask client\n",
    "\n",
    "- In order to work with a Dask cluster we need a `Client` object.\n",
    "- It represents the connection to that cluster and allows to configure execution-related parameters (e.g. number of cores, memory). \n",
    "- The client object is just the intermediary between our client session and the cluster resources, \n",
    "- Dask supports many different resource managers.\n",
    "- We will follow the [Dask documentation](https://distributed.dask.org/en/stable/client.html) regarding the creation of a `Client`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4e156e5",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from dask.distributed import Client, LocalCluster\n",
    "cluster = LocalCluster(n_workers=2, threads_per_worker=1, processes=True, memory_limit=\"2GiB\")\n",
    "client = Client(cluster)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "601ffed1",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Create a ROOT dataframe\n",
    "\n",
    "We now create an RDataFrame based on the same dataset seen in the exercise [rdataframe-dimuon](exercises/rdataframe-dimuon.ipynb).\n",
    "\n",
    "A Dask `RDataFrame` receives two extra parameters: \n",
    "- the number of partitions to apply to the dataset (`npartitions`)\n",
    "- the `Client` object (`daskclient`). \n",
    "\n",
    "Besides this detail, a Dask `RDataFrame` is not different from a local `RDataFrame`: the analysis presented in this notebook would not change if we wanted to execute it locally."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "405cf12f",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Use a Dask RDataFrame\n",
    "RDataFrame = ROOT.RDF.Experimental.Distributed.Dask.RDataFrame\n",
    "\n",
    "df = RDataFrame(\"h42\",\n",
    "                \"https://root.cern/files/h1big.root\",\n",
    "                npartitions=4,\n",
    "                daskclient=client)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "563a28e4",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Run your analysis unchanged\n",
    "\n",
    "- From now on, the rest of your application can be written **exactly** as we have seen with local RDataFrame. \n",
    "\n",
    "- The goal of the distributed RDataFrame module is to support all the traditional RDataFrame operations (those that make sense in a distributed context at least). \n",
    "\n",
    "- Currently only a subset of those is available and can be found in the corresponding [section of the documentation](https://root.cern/doc/master/classROOT_1_1RDataFrame.html#distrdf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "43562f8d",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "df1 = df.Filter(\"nevent > 1\")\n",
    "df2 = df1.Define(\"mpt\",\"sqrt(xpt*xpt + ypt*ypt)\")\n",
    "c = df.Count()\n",
    "m = df2.Mean(\"mpt\")\n",
    "print(f\"Number of events after processing: {c.GetValue()}\")\n",
    "print(f\"Mean of column 'mpt': {m.GetValue()}\")"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}