Merge pull request #254 from skmatti/cudf-docs-examples

Add docs and examples for cudf support

Author: martindurant

docs/source/collections.rst

@@ -6,7 +6,8 @@ bring special consideration to certain types of data:
 
 1.  ``streamz.batch``: supports streams of lists of Python objects like tuples
     or dictionaries
-2.  ``streamz.dataframe``: supports streams of Pandas dataframes or Pandas series
+2.  ``streamz.dataframe``: supports streams of Pandas/cudf dataframes or Pandas/cudf series.
+    cudf support is in beta phase and has limited functionality as of cudf version ``0.8``
 
 These high-level APIs help us handle common situations in data processing.
 They help us implement complex algorithms and also improve efficiency.

docs/source/gpu-dataframes.rst

@@ -0,0 +1,41 @@
+Streaming GPU DataFrames(cudf)
+------------------------------
+
+The ``streamz.dataframe`` module provides DataFrame-like interface on streaming
+data as described in ``dataframes`` documentation. It provides support for dataframe
+like libraries such as pandas and cudf. This documentation is specific to streaming GPU
+dataframes(cudf).
+
+The example in the ``dataframes`` documentation is rewritten below using cudf dataframes
+just by replacing ``pandas`` module with ``cudf``:
+
+.. code-block:: python
+
+   import cudf
+   from streamz.dataframe import DataFrame
+
+   example = cudf.DataFrame({'name': [], 'amount': []})
+   sdf = DataFrame(stream, example=example)
+
+   sdf[sdf.name == 'Alice'].amount.sum()
+
+
+Supported Operations
+--------------------
+
+Streaming cudf dataframes support the following classes of operations
+
+-  Elementwise operations like ``df.x + 1``
+-  Filtering like ``df[df.name == 'Alice']``
+-  Column addition like ``df['z'] = df.x + df.y``
+-  Reductions like ``df.amount.mean()``
+-  Windowed aggregations (fixed length) like ``df.window(n=100).amount.sum()``
+
+The following operations are not supported with cudf(as of version 0.8) yet
+-  Groupby-aggregations like ``df.groupby(df.name).amount.mean()``
+-  Windowed aggregations (index valued) like ``df.window(value='2h').amount.sum()``
+-  Windowed groupby aggregations like ``df.window(value='2h').groupby('name').amount.sum()``
+
+
+Window based Aggregations with cudf are supported just as explained in ``dataframes`` documentation.
+The support for groupby operations will be added in future.

examples/gpu-dataframes.ipynb

@@ -0,0 +1,125 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from streamz.dataframe import DataFrame\n",
+    "import cudf"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Basic example"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\n",
+    "cu_df = cudf.DataFrame({'x': np.arange(10, dtype=float)+10, 'y': [1.0, 2.0] * 5})\n",
+    "\n",
+    "sdf = DataFrame(example=cu_df)\n",
+    "\n",
+    "L = sdf.window(n=15).x.sum().stream.sink_to_list()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "sdf.emit(cu_df.iloc[:8])\n",
+    "sdf.emit(cu_df)\n",
+    "sdf.emit(cu_df)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "print(L[0])\n",
+    "print(L[1])\n",
+    "print(L[2])"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Advanced example\n",
+    "The following pipeline reads json encoded strings from Kafka in batches and process them on GPUs and write the result back to a different Kafka topic. This pipeline can be easily extended to run on Dask Stream as well.\n",
+    "Note: Uses cudf 0.8"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# read messages from kafka and create a stream\n",
+    "\n",
+    "consume_topic = \"my-topic\"\n",
+    "produce_topic = \"my-out-topic\"\n",
+    "bootstrap_servers = 'localhost:9092'\n",
+    "consumer_conf = {'bootstrap.servers': bootstrap_servers,\n",
+    "                 'group.id': 'group-123', 'session.timeout.ms': 600}\n",
+    "producer_conf = {'bootstrap.servers': bootstrap_servers}\n",
+    "\n",
+    "stream = Stream.from_kafka_batched(consume_topic, consumer_conf, poll_interval='10s',\n",
+    "                                   npartitions=10, asynchronous=True)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# convert batch of encoded json strings to gpu dataframes\n",
+    "cudf_stream = stream\\\n",
+    "                    .map(lambda msgs: \"\\n\".join([msg.decode('utf-8') for msg in msgs]))\\\n",
+    "                    .map(cudf.read_json, lines=True)\n",
+    "\n",
+    "# create a streamz dataframe from the above stream and sample dataframe\n",
+    "cudf_example = cudf.DataFrame({'x': np.arange(10, dtype=float)+10, 'y': [1.0, 2.0] * 5})\n",
+    "stdf = DataFrame(cudf_stream, example=cudf_example)\n",
+    "\n",
+    "# perform aggregation and write to kafka\n",
+    "stdf.window(n=15).x.mean().stream.to_kafka(produce_topic, producer_conf)\n"
+   ]
+  }
+ ],
+ "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.3"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}