notebooks/services/kafka.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "189f5004",
   "metadata": {},
   "source": [
    "## Kafka API"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5c845aa5",
   "metadata": {},
   "source": [
    "This API provides the possibility to manage schemas, topics and setting up the connection to the kafka brokers."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d6de58bc",
   "metadata": {},
   "source": [
    "## Scope"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "279bcdc9",
   "metadata": {},
   "source": [
    "* Create a schema\n",
    "* Create a topic\n",
    "* Produce messages to the broker\n",
    "* Consume messages from the broker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "19215d11",
   "metadata": {},
   "outputs": [],
   "source": [
    "import hopsworks"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "13f50781",
   "metadata": {},
   "source": [
    "## Connect to the cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "ac36bf2e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connected. Call `.close()` to terminate connection gracefully.\n"
     ]
    }
   ],
   "source": [
    "# Connect to your cluster, to be used running inside Jupyter or jobs inside the cluster.\n",
    "connection = hopsworks.connection()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e547e286",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Uncomment when connecting to the cluster from an external environment.\n",
    "# connection = hopsworks.connection(project='my_project', host='my_instance', port=443, api_key_value='apikey')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a0c2440",
   "metadata": {},
   "source": [
    "## Get Project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "b5deb653",
   "metadata": {},
   "outputs": [],
   "source": [
    "project = connection.get_project()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f98c8e4b",
   "metadata": {},
   "source": [
    "## Get the API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "a288cc51",
   "metadata": {},
   "outputs": [],
   "source": [
    "kafka_api = project.get_kafka_api()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1c66fa2f",
   "metadata": {},
   "source": [
    "## Define an avro schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "f3daf802",
   "metadata": {},
   "outputs": [],
   "source": [
    "SCHEMA_NAME=\"schema_example\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "db6f3fdb",
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = {\n",
    "  \"type\": \"record\",\n",
    "  \"name\": \"tutorial\",\n",
    "  \"fields\": [\n",
    "    {\n",
    "      \"name\": \"id\",\n",
    "      \"type\": \"int\"  \n",
    "    },\n",
    "    {\n",
    "      \"name\": \"data\",\n",
    "      \"type\": \"string\"\n",
    "    }\n",
    "  ]\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "bb485296",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "KafkaSchema('schema_example', 1)"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_schema = kafka_api.create_schema(SCHEMA_NAME, schema)\n",
    "my_schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "dc5c3152",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'schema_example'"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_schema.subject"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "fdc98cfb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_schema.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "e124116c",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_schema.schema"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "41d92043",
   "metadata": {},
   "source": [
    "## Define a topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "83bb5ba5",
   "metadata": {},
   "outputs": [],
   "source": [
    "TOPIC_NAME=\"topic_example\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "48a41c94",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "KafkaTopic('topic_example')"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_topic = kafka_api.create_topic(TOPIC_NAME, SCHEMA_NAME, 1, replicas=1, partitions=1)\n",
    "my_topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "7647669c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'topic_example'"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_topic.name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "56e1eaa1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_topic.replicas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "a6c2a0e9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_topic.partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "f37dc434",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "KafkaSchema('schema_example', 1)"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_topic.schema"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "17c1affb",
   "metadata": {},
   "source": [
    "## Produce messages to topic\n",
    "#### Currently _kafka_api.get_default_config()_ is only supported for use in a notebook or job inside Hopsworks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "7218c882",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'bootstrap.servers': '10.0.2.15:9091',\n",
       " 'security.protocol': 'SSL',\n",
       " 'ssl.ca.location': 'ca_chain.pem',\n",
       " 'ssl.certificate.location': 'client.pem',\n",
       " 'ssl.key.location': 'client_key.pem',\n",
       " 'group.id': 'my-group-id'}"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "producer_config = kafka_api.get_default_config()\n",
    "producer_config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "31dad3b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from confluent_kafka import Producer, Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "31c73b42",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Configure producer\n",
    "producer = Producer(producer_config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "e222a6c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def delivery_callback(err, msg):\n",
    "    \"\"\"\n",
    "    Optional per-message delivery callback (triggered by poll() or flush())\n",
    "    when a message has been successfully delivered or permanently\n",
    "    failed delivery (after retries).\n",
    "    \"\"\"\n",
    "    if err:\n",
    "        print(\"Message failed delivery: {}\".format(err))\n",
    "    else:\n",
    "        print('Message: {} delivered to topic: {}, partition: {}, offset: {}, timestamp: {}'.format(msg.value(), msg.topic(), msg.partition(), msg.offset(), msg.timestamp()))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "727c6266",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "DeprecationWarning: PY_SSIZE_T_CLEAN will be required for '#' formats\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Message: b'{\"id\": 0, \"data\": \"36039d5a-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 0, timestamp: (1, 1650887990722)\n",
      "Message: b'{\"id\": 1, \"data\": \"3603b9fc-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 1, timestamp: (1, 1650887990722)\n",
      "Message: b'{\"id\": 2, \"data\": \"3603ceb0-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 2, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 3, \"data\": \"3603d41e-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 3, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 4, \"data\": \"3603dd38-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 4, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 5, \"data\": \"3603e134-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 5, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 6, \"data\": \"3603e4fe-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 6, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 7, \"data\": \"3603e8a0-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 7, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 8, \"data\": \"3603ec56-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 8, timestamp: (1, 1650887990723)\n",
      "Message: b'{\"id\": 9, \"data\": \"3603ef94-c48f-11ec-af74-080027fffec3\"}' delivered to topic: topic_example, partition: 0, offset: 9, timestamp: (1, 1650887990723)\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import uuid\n",
    "import json\n",
    "for i in range(0, 10):\n",
    "    producer.produce(TOPIC_NAME, json.dumps({\"id\": i, \"data\": str(uuid.uuid1())}), \"key\", callback=delivery_callback)\n",
    "    \n",
    "# Trigger the sending of all messages to the brokers, 20sec timeout\n",
    "producer.flush(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "03ae69b1",
   "metadata": {},
   "source": [
    "## Consume messages from topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "893377d0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'bootstrap.servers': '10.0.2.15:9091',\n",
       " 'security.protocol': 'SSL',\n",
       " 'ssl.ca.location': 'ca_chain.pem',\n",
       " 'ssl.certificate.location': 'client.pem',\n",
       " 'ssl.key.location': 'client_key.pem',\n",
       " 'group.id': 'my-group-id',\n",
       " 'default.topic.config': {'auto.offset.reset': 'earliest'}}"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from confluent_kafka import Consumer\n",
    "\n",
    "consumer_config = kafka_api.get_default_config()\n",
    "consumer_config['default.topic.config'] = {'auto.offset.reset': 'earliest'}\n",
    "consumer_config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "e81a1ae4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Configure consumer\n",
    "consumer = Consumer(consumer_config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "7a036a68",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Subscribe to topic\n",
    "consumer.subscribe([TOPIC_NAME])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "479be4c8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b'{\"id\": 0, \"data\": \"36039d5a-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 1, \"data\": \"3603b9fc-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 2, \"data\": \"3603ceb0-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 3, \"data\": \"3603d41e-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 4, \"data\": \"3603dd38-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 5, \"data\": \"3603e134-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 6, \"data\": \"3603e4fe-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 7, \"data\": \"3603e8a0-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 8, \"data\": \"3603ec56-c48f-11ec-af74-080027fffec3\"}'\n",
      "b'{\"id\": 9, \"data\": \"3603ef94-c48f-11ec-af74-080027fffec3\"}'\n"
     ]
    }
   ],
   "source": [
    "for i in range(0, 10):\n",
    "    msg = consumer.poll(timeout=30.0)\n",
    "    print(msg.value())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "bac0f150",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_topic.delete()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "3cd0e106",
   "metadata": {},
   "outputs": [],
   "source": [
    "my_schema.delete()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python",
   "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.8.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}