diff --git a/examples/gpu-dataframes.ipynb b/examples/gpu-dataframes.ipynb new file mode 100644 index 00000000..89b61c58 --- /dev/null +++ b/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 +}