Add examples for cudf

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
+}