Streams-BasicSample.ipynb

{"cells": [{"metadata": {}, "cell_type": "markdown", "source": "# Basic Streams Sample\n\nThe basic cells needed are included below.  Make any changes and add your sources, analytics and outputs.\n\nIn this notebook, you'll see examples of how to:\n 1. [Setup your Streams instance](#setup)\n 1. [Create the application](#create)\n 1. [Submit the application](#launch)\n 1. [Connect to the running application to view data](#view)\n 1. [Stop the application](#cancel)\n\n# Overview\n\n**Introduction**\n\nThis sample shows how to create and launch an application.\n\n**How it works**\n   \nThe Python application created in this notebook is submitted to the IBM Streams service for execution.\n\n<img src=\"https://developer.ibm.com/streamsdev/wp-content/uploads/sites/15/2019/04/how-it-works.jpg\" alt=\"How it works\">\n\n\n## Documentation\n    \n- [Streams Python development guide](https://ibmstreams.github.io/streamsx.documentation/docs/latest/python/)\n- [Streams Python API](https://streamsxtopology.readthedocs.io/)\n\n\n\n## <a name=\"setup\"> </a> 1. Setup\n\n### 1.1 Add credentials for the IBM Streams service\n\nIn order to submit a Streams application you need to provide the name of the Streams instance.\n\n1. From the navigation menu, click **My instances**.\n2. Click the **Provisioned Instances** tab.\n3. Update the value of `streams_instance_name` in the cell below according to your Streams instance name.\n"}, {"metadata": {}, "cell_type": "code", "source": "## Paste snippet from development guide\nfrom streamsx.topology.context import ConfigParams\nfrom streamsx.topology import context\nimport json\nimport getpass\n        \n        \nservice_cfg  = {}\n        \nSA_credentials=getpass.getpass('Streaming Analytics credentials:')\nservice_cfg[ConfigParams.SERVICE_DEFINITION] = json.loads(SA_credentials)\n\ndef submit_topology(topo):\n    global service_cfg\n    service_cfg[context.ConfigParams.SSL_VERIFY] = False\n\n    # This specifies how the application will be deployed\n\n    contextType = context.ContextTypes.STREAMING_ANALYTICS_SERVICE\n\n    return context.submit (contextType, topo, config = service_cfg) ", "execution_count": 2, "outputs": [{"output_type": "stream", "name": "stdout", "text": "Streaming Analytics credentials:\u00b7\u00b7\u00b7\u00b7\u00b7\u00b7\u00b7\u00b7\n"}]}, {"metadata": {}, "cell_type": "markdown", "source": "### 1.2 Import  the `streamsx` package and verify the package version"}, {"metadata": {}, "cell_type": "code", "source": "import streamsx.topology.context\nprint(\"INFO: streamsx package version: \" + streamsx.topology.context.__version__)\n\n#For more details uncomment line below.\n#!pip show streamsx", "execution_count": 3, "outputs": [{"output_type": "stream", "text": "INFO: streamsx package version: 1.13.14\n", "name": "stdout"}]}, {"metadata": {}, "cell_type": "markdown", "source": "<a id=\"create\"></a>\n# 2. Create the application"}, {"metadata": {}, "cell_type": "code", "source": "#create topology...change the name to something more meaningful\nfrom streamsx.topology.topology import Topology\nimport streamsx.topology.context\n\ntopo = Topology(name=\"BasicTemplate\")", "execution_count": 4, "outputs": [{"output_type": "stream", "text": "2019-12-23 15:32:44,836 - __PROJECT_LIB__ - ERROR - ProjectHandle: Project ID neither provided nor found in the environment.\n", "name": "stderr"}]}, {"metadata": {}, "cell_type": "code", "source": "# add stream...delete and replace this with your streams and analytics\n# this is added so the topology will run (needs a source)\n# produces a random integer between 1 and 100 every tenth of a second\nimport random, time\ndef readings():\n    while True:\n        time.sleep(0.1)\n        yield random.randint(1,100)\n\nsource = topo.source(readings)\n\n# create a view to watch the stream data while running\nsource_view = source.view(name=\"RandomIntegers\", description=\"Random integers from source stream\")", "execution_count": 5, "outputs": []}, {"metadata": {}, "cell_type": "markdown", "source": "<a id=\"launch\"></a>\n# 3. Submit the application\nA running Streams application is called a *job*. By submitting the topology we create a job."}, {"metadata": {}, "cell_type": "code", "source": "#Submit\nprint(\"Submitting Topology to Streams for execution..\")\nsubmission_result = submit_topology(topo)\n\nif submission_result.job:\n    streams_job = submission_result.job\n    print (\"JobId: \", streams_job.id , \"\\nJob name: \", streams_job.name)\nelse:\n    print(\"Submission failed: \"   + str(submssion_result))", "execution_count": 9, "outputs": [{"output_type": "stream", "text": "Submitting Topology to Streams for execution..\n", "name": "stdout"}, {"output_type": "display_data", "data": {"text/plain": "IntProgress(value=0, bar_style='info', description='Initializing', max=10, style=ProgressStyle(description_wid\u2026", "application/vnd.jupyter.widget-view+json": {"version_major": 2, "version_minor": 0, "model_id": "2722865627e047f18ce39e122cc235ff"}}, "metadata": {}}, {"output_type": "stream", "text": "JobId:  2 \nJob name:  notebook::BasicTemplate_2\n", "name": "stdout"}]}, {"metadata": {}, "cell_type": "markdown", "source": "<a name=\"view\"></a>\n# 4. Use a `View` to access data from the job\nNow that the job is started, use the `View` object to start retrieving data from a `Stream`."}, {"metadata": {}, "cell_type": "code", "source": "#show 30 seconds of data\nif submission_result.job:\n    queue = source_view.start_data_fetch()\n    timeSpan = time.time() + 30;\n    while time.time() < timeSpan:\n        print(queue.get())\n    source_view.stop_data_fetch()\n", "execution_count": 10, "outputs": [{"output_type": "stream", "text": "59\n92\n30\n68\n85\n70\n73\n26\n21\n63\n54\n43\n49\n53\n78\n16\n46\n67\n58\n66\n9\n53\n18\n32\n51\n94\n42\n32\n31\n65\n62\n10\n48\n81\n56\n9\n93\n1\n98\n25\n89\n28\n18\n68\n17\n33\n95\n84\n24\n71\n15\n51\n7\n76\n77\n11\n92\n32\n56\n11\n86\n12\n99\n40\n86\n3\n8\n72\n37\n43\n74\n94\n49\n29\n93\n37\n100\n15\n8\n100\n60\n60\n92\n73\n73\n70\n80\n75\n24\n76\n69\n11\n13\n87\n17\n68\n38\n79\n62\n56\n6\n27\n31\n25\n28\n96\n1\n2\n46\n33\n100\n15\n55\n1\n91\n47\n43\n18\n25\n40\n13\n48\n53\n91\n46\n63\n8\n83\n80\n92\n80\n12\n7\n38\n42\n15\n64\n83\n93\n36\n41\n80\n65\n42\n88\n68\n72\n55\n90\n30\n15\n73\n13\n100\n60\n93\n99\n98\n43\n85\n60\n27\n49\n71\n19\n68\n42\n81\n74\n73\n48\n71\n40\n72\n66\n21\n23\n10\n79\n12\n96\n17\n8\n53\n1\n43\n55\n22\n81\n21\n93\n87\n57\n94\n23\n34\n44\n19\n98\n45\n53\n44\n2\n5\n95\n54\n92\n73\n36\n42\n6\n79\n52\n57\n22\n73\n65\n34\n19\n34\n24\n26\n81\n1\n93\n42\n43\n69\n50\n34\n9\n18\n29\n59\n50\n11\n61\n72\n56\n7\n51\n69\n32\n42\n28\n52\n13\n94\n82\n30\n63\n82\n17\n48\n6\n6\n33\n17\n32\n92\n15\n8\n81\n55\n10\n76\n56\n5\n98\n42\n6\n79\n6\n15\n29\n41\n5\n2\n67\n90\n91\n35\n90\n37\n17\n63\n37\n", "name": "stdout"}]}, {"metadata": {}, "cell_type": "markdown", "source": "\n<a name=\"cancel\"></a>\n\n# 5. Cancel the job\n\nThis cell generates widgets you can use to cancel the jobs."}, {"metadata": {}, "cell_type": "code", "source": "submission_result.cancel_job_button()", "execution_count": 11, "outputs": [{"output_type": "display_data", "data": {"text/plain": "VBox(children=(Button(button_style='danger', description='Cancel job: notebook::BasicTemplate_2', layout=Layou\u2026", "application/vnd.jupyter.widget-view+json": {"version_major": 2, "version_minor": 0, "model_id": "ab8ed3d8d1784ff6bd49d2f1d15d6197"}}, "metadata": {}}]}, {"metadata": {}, "cell_type": "markdown", "source": "You can also interact with the job through the [Job](https://streamsxtopology.readthedocs.io/en/stable/streamsx.rest_primitives.html#streamsx.rest_primitives.Job) object returned from `submission_result.job`\n\nFor example, use `streams_job.cancel()` to cancel the running job directly."}], "metadata": {"kernelspec": {"name": "python3", "display_name": "Python 3.6", "language": "python"}, "language_info": {"name": "python", "version": "3.6.8", "mimetype": "text/x-python", "codemirror_mode": {"name": "ipython", "version": 3}, "pygments_lexer": "ipython3", "nbconvert_exporter": "python", "file_extension": ".py"}}, "nbformat": 4, "nbformat_minor": 1}