kafkaconsumer1.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/aomidvar/scrapper-price-comparison/blob/main/kafkaconsumer1.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "m8YpeHnHwSG0",
        "outputId": "fff8e174-1b95-42f4-869f-46f8db3cc32b"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "openjdk version \"11.0.19\" 2023-04-18\n",
            "OpenJDK Runtime Environment (build 11.0.19+7-post-Ubuntu-0ubuntu120.04.1)\n",
            "OpenJDK 64-Bit Server VM (build 11.0.19+7-post-Ubuntu-0ubuntu120.04.1, mixed mode, sharing)\n"
          ]
        }
      ],
      "source": [
        "!java -version"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "id": "260K5pW4zTK9",
        "outputId": "d578ba8e-1494-47ed-ab63-442a959cf1cf"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting apache-flink[kafka]\n",
            "  Downloading apache_flink-1.17.0-cp310-cp310-manylinux1_x86_64.whl (5.7 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m5.7/5.7 MB\u001b[0m \u001b[31m38.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h\u001b[33mWARNING: apache-flink 1.17.0 does not provide the extra 'kafka'\u001b[0m\u001b[33m\n",
            "\u001b[0mRequirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from apache-flink[kafka]) (0.10.9.7)\n",
            "Requirement already satisfied: python-dateutil<3,>=2.8.0 in /usr/local/lib/python3.10/dist-packages (from apache-flink[kafka]) (2.8.2)\n",
            "Collecting apache-beam==2.43.0 (from apache-flink[kafka])\n",
            "  Downloading apache_beam-2.43.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (14.0 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m14.0/14.0 MB\u001b[0m \u001b[31m75.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting cloudpickle==2.2.0 (from apache-flink[kafka])\n",
            "  Downloading cloudpickle-2.2.0-py3-none-any.whl (25 kB)\n",
            "Collecting avro-python3!=1.9.2,<1.10.0,>=1.8.1 (from apache-flink[kafka])\n",
            "  Downloading avro-python3-1.9.2.1.tar.gz (37 kB)\n",
            "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: pytz>=2018.3 in /usr/local/lib/python3.10/dist-packages (from apache-flink[kafka]) (2022.7.1)\n",
            "Collecting fastavro<1.4.8,>=1.1.0 (from apache-flink[kafka])\n",
            "  Downloading fastavro-1.4.7-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (2.6 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m2.6/2.6 MB\u001b[0m \u001b[31m85.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: requests>=2.26.0 in /usr/local/lib/python3.10/dist-packages (from apache-flink[kafka]) (2.27.1)\n",
            "Requirement already satisfied: protobuf<=3.21,>=3.19.0 in /usr/local/lib/python3.10/dist-packages (from apache-flink[kafka]) (3.20.3)\n",
            "Collecting numpy<1.22.0,>=1.21.4 (from apache-flink[kafka])\n",
            "  Downloading numpy-1.21.6-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (15.9 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m15.9/15.9 MB\u001b[0m \u001b[31m77.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting pandas<1.4.0,>=1.3.0 (from apache-flink[kafka])\n",
            "  Downloading pandas-1.3.5-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (11.5 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m11.5/11.5 MB\u001b[0m \u001b[31m98.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting pyarrow<9.0.0,>=5.0.0 (from apache-flink[kafka])\n",
            "  Downloading pyarrow-8.0.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (29.4 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m29.4/29.4 MB\u001b[0m \u001b[31m15.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting httplib2<=0.20.4,>=0.19.0 (from apache-flink[kafka])\n",
            "  Downloading httplib2-0.20.4-py3-none-any.whl (96 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m96.6/96.6 kB\u001b[0m \u001b[31m10.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting apache-flink-libraries<1.17.1,>=1.17.0 (from apache-flink[kafka])\n",
            "  Downloading apache-flink-libraries-1.17.0.tar.gz (240.0 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m240.0/240.0 MB\u001b[0m \u001b[31m2.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Collecting pemja==0.3.0 (from apache-flink[kafka])\n",
            "  Downloading pemja-0.3.0-cp310-cp310-manylinux1_x86_64.whl (332 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m332.7/332.7 kB\u001b[0m \u001b[31m28.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting crcmod<2.0,>=1.7 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading crcmod-1.7.tar.gz (89 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m89.7/89.7 kB\u001b[0m \u001b[31m9.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Collecting orjson<4.0 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading orjson-3.8.12-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (137 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m137.2/137.2 kB\u001b[0m \u001b[31m13.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting dill<0.3.2,>=0.3.1.1 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading dill-0.3.1.1.tar.gz (151 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m152.0/152.0 kB\u001b[0m \u001b[31m14.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Collecting fasteners<1.0,>=0.3 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading fasteners-0.18-py3-none-any.whl (18 kB)\n",
            "Requirement already satisfied: grpcio!=1.48.0,<2,>=1.33.1 in /usr/local/lib/python3.10/dist-packages (from apache-beam==2.43.0->apache-flink[kafka]) (1.54.0)\n",
            "Collecting hdfs<3.0.0,>=2.1.0 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading hdfs-2.7.0-py3-none-any.whl (34 kB)\n",
            "Collecting objsize<0.6.0,>=0.5.2 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading objsize-0.5.2-py3-none-any.whl (8.2 kB)\n",
            "Collecting pymongo<4.0.0,>=3.8.0 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading pymongo-3.13.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (516 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m516.2/516.2 kB\u001b[0m \u001b[31m35.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: proto-plus<2,>=1.7.1 in /usr/local/lib/python3.10/dist-packages (from apache-beam==2.43.0->apache-flink[kafka]) (1.22.2)\n",
            "Requirement already satisfied: pydot<2,>=1.2.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam==2.43.0->apache-flink[kafka]) (1.4.2)\n",
            "Requirement already satisfied: regex>=2020.6.8 in /usr/local/lib/python3.10/dist-packages (from apache-beam==2.43.0->apache-flink[kafka]) (2022.10.31)\n",
            "Requirement already satisfied: typing-extensions>=3.7.0 in /usr/local/lib/python3.10/dist-packages (from apache-beam==2.43.0->apache-flink[kafka]) (4.5.0)\n",
            "Collecting zstandard<1,>=0.18.0 (from apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading zstandard-0.21.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (2.7 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m2.7/2.7 MB\u001b[0m \u001b[31m62.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting find-libpython (from pemja==0.3.0->apache-flink[kafka])\n",
            "  Downloading find_libpython-0.3.0-py3-none-any.whl (8.5 kB)\n",
            "Requirement already satisfied: pyparsing!=3.0.0,!=3.0.1,!=3.0.2,!=3.0.3,<4,>=2.4.2 in /usr/local/lib/python3.10/dist-packages (from httplib2<=0.20.4,>=0.19.0->apache-flink[kafka]) (3.0.9)\n",
            "Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.10/dist-packages (from python-dateutil<3,>=2.8.0->apache-flink[kafka]) (1.16.0)\n",
            "Requirement already satisfied: urllib3<1.27,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests>=2.26.0->apache-flink[kafka]) (1.26.15)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests>=2.26.0->apache-flink[kafka]) (2022.12.7)\n",
            "Requirement already satisfied: charset-normalizer~=2.0.0 in /usr/local/lib/python3.10/dist-packages (from requests>=2.26.0->apache-flink[kafka]) (2.0.12)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests>=2.26.0->apache-flink[kafka]) (3.4)\n",
            "Collecting docopt (from hdfs<3.0.0,>=2.1.0->apache-beam==2.43.0->apache-flink[kafka])\n",
            "  Downloading docopt-0.6.2.tar.gz (25 kB)\n",
            "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Building wheels for collected packages: apache-flink-libraries, avro-python3, crcmod, dill, docopt\n",
            "  Building wheel for apache-flink-libraries (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for apache-flink-libraries: filename=apache_flink_libraries-1.17.0-py2.py3-none-any.whl size=240040471 sha256=67d4a1641cfc62ded4dec49015ae536a89a7d23941c5317fc7613cc2a4bf627b\n",
            "  Stored in directory: /root/.cache/pip/wheels/aa/f3/9f/f268297890997d81b90bbd0f2664d64fbd7c94dab680baab28\n",
            "  Building wheel for avro-python3 (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for avro-python3: filename=avro_python3-1.9.2.1-py3-none-any.whl size=43497 sha256=2f8d6f73e4cd2f819ba7d4d12494ecb1bdf429fba8c5a7f9b12af76d7228893d\n",
            "  Stored in directory: /root/.cache/pip/wheels/f9/dc/61/0e621c0c1c1d61c4699d6c6565de69171cbec6f244b8b380f5\n",
            "  Building wheel for crcmod (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for crcmod: filename=crcmod-1.7-cp310-cp310-linux_x86_64.whl size=37097 sha256=cdce97f5d19bc52daf3722dc5e4471265c751d8c71f8cbcc4dd23cdb75e0cc0d\n",
            "  Stored in directory: /root/.cache/pip/wheels/85/4c/07/72215c529bd59d67e3dac29711d7aba1b692f543c808ba9e86\n",
            "  Building wheel for dill (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for dill: filename=dill-0.3.1.1-py3-none-any.whl size=78545 sha256=e3642da58ecc3b31691b5d243df7a708248b17fa8fe98ea4a95092c75bb9c2fb\n",
            "  Stored in directory: /root/.cache/pip/wheels/ea/e2/86/64980d90e297e7bf2ce588c2b96e818f5399c515c4bb8a7e4f\n",
            "  Building wheel for docopt (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for docopt: filename=docopt-0.6.2-py2.py3-none-any.whl size=13707 sha256=53c9ae76e51f1279b3c84337bce5aa934b25a1db831058b5f37c80c92dceaaee\n",
            "  Stored in directory: /root/.cache/pip/wheels/fc/ab/d4/5da2067ac95b36618c629a5f93f809425700506f72c9732fac\n",
            "Successfully built apache-flink-libraries avro-python3 crcmod dill docopt\n",
            "Installing collected packages: find-libpython, docopt, crcmod, zstandard, pymongo, pemja, orjson, objsize, numpy, httplib2, fasteners, fastavro, dill, cloudpickle, avro-python3, apache-flink-libraries, pyarrow, pandas, hdfs, apache-beam, apache-flink\n",
            "  Attempting uninstall: numpy\n",
            "    Found existing installation: numpy 1.22.4\n",
            "    Uninstalling numpy-1.22.4:\n",
            "      Successfully uninstalled numpy-1.22.4\n",
            "  Attempting uninstall: httplib2\n",
            "    Found existing installation: httplib2 0.21.0\n",
            "    Uninstalling httplib2-0.21.0:\n",
            "      Successfully uninstalled httplib2-0.21.0\n",
            "  Attempting uninstall: cloudpickle\n",
            "    Found existing installation: cloudpickle 2.2.1\n",
            "    Uninstalling cloudpickle-2.2.1:\n",
            "      Successfully uninstalled cloudpickle-2.2.1\n",
            "  Attempting uninstall: pyarrow\n",
            "    Found existing installation: pyarrow 9.0.0\n",
            "    Uninstalling pyarrow-9.0.0:\n",
            "      Successfully uninstalled pyarrow-9.0.0\n",
            "  Attempting uninstall: pandas\n",
            "    Found existing installation: pandas 1.5.3\n",
            "    Uninstalling pandas-1.5.3:\n",
            "      Successfully uninstalled pandas-1.5.3\n",
            "\u001b[31mERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.\n",
            "google-colab 1.0.0 requires pandas==1.5.3, but you have pandas 1.3.5 which is incompatible.\n",
            "tensorflow 2.12.0 requires numpy<1.24,>=1.22, but you have numpy 1.21.6 which is incompatible.\u001b[0m\u001b[31m\n",
            "\u001b[0mSuccessfully installed apache-beam-2.43.0 apache-flink-1.17.0 apache-flink-libraries-1.17.0 avro-python3-1.9.2.1 cloudpickle-2.2.0 crcmod-1.7 dill-0.3.1.1 docopt-0.6.2 fastavro-1.4.7 fasteners-0.18 find-libpython-0.3.0 hdfs-2.7.0 httplib2-0.20.4 numpy-1.21.6 objsize-0.5.2 orjson-3.8.12 pandas-1.3.5 pemja-0.3.0 pyarrow-8.0.0 pymongo-3.13.0 zstandard-0.21.0\n"
          ]
        },
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.colab-display-data+json": {
              "pip_warning": {
                "packages": [
                  "httplib2",
                  "numpy"
                ]
              }
            }
          },
          "metadata": {}
        }
      ],
      "source": [
        "pip install apache-flink[kafka]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "lzkNw5bzFGLv",
        "outputId": "44835028-6989-4346-99f1-399a980f8b3b"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting kafka-python\n",
            "  Downloading kafka_python-2.0.2-py2.py3-none-any.whl (246 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m246.5/246.5 kB\u001b[0m \u001b[31m5.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hInstalling collected packages: kafka-python\n",
            "Successfully installed kafka-python-2.0.2\n"
          ]
        }
      ],
      "source": [
        "pip install kafka-python"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "rnmugNfW3THp",
        "outputId": "a37b5320-fa02-4c46-b179-971080917cda"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting confluent-kafka\n",
            "  Downloading confluent_kafka-2.1.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (3.9 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m3.9/3.9 MB\u001b[0m \u001b[31m33.4 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hInstalling collected packages: confluent-kafka\n",
            "Successfully installed confluent-kafka-2.1.1\n"
          ]
        }
      ],
      "source": [
        "pip install confluent-kafka"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 74
        },
        "id": "CRHPglRl3cff",
        "outputId": "08add3e8-91e2-4c48-95ca-807fd2d056c2"
      },
      "outputs": [
        {
          "data": {
            "text/html": [
              "\n",
              "     <input type=\"file\" id=\"files-f9263790-564f-4339-b241-b071e53a6bab\" name=\"files[]\" multiple disabled\n",
              "        style=\"border:none\" />\n",
              "     <output id=\"result-f9263790-564f-4339-b241-b071e53a6bab\">\n",
              "      Upload widget is only available when the cell has been executed in the\n",
              "      current browser session. Please rerun this cell to enable.\n",
              "      </output>\n",
              "      <script>// Copyright 2017 Google LLC\n",
              "//\n",
              "// Licensed under the Apache License, Version 2.0 (the \"License\");\n",
              "// you may not use this file except in compliance with the License.\n",
              "// You may obtain a copy of the License at\n",
              "//\n",
              "//      http://www.apache.org/licenses/LICENSE-2.0\n",
              "//\n",
              "// Unless required by applicable law or agreed to in writing, software\n",
              "// distributed under the License is distributed on an \"AS IS\" BASIS,\n",
              "// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
              "// See the License for the specific language governing permissions and\n",
              "// limitations under the License.\n",
              "\n",
              "/**\n",
              " * @fileoverview Helpers for google.colab Python module.\n",
              " */\n",
              "(function(scope) {\n",
              "function span(text, styleAttributes = {}) {\n",
              "  const element = document.createElement('span');\n",
              "  element.textContent = text;\n",
              "  for (const key of Object.keys(styleAttributes)) {\n",
              "    element.style[key] = styleAttributes[key];\n",
              "  }\n",
              "  return element;\n",
              "}\n",
              "\n",
              "// Max number of bytes which will be uploaded at a time.\n",
              "const MAX_PAYLOAD_SIZE = 100 * 1024;\n",
              "\n",
              "function _uploadFiles(inputId, outputId) {\n",
              "  const steps = uploadFilesStep(inputId, outputId);\n",
              "  const outputElement = document.getElementById(outputId);\n",
              "  // Cache steps on the outputElement to make it available for the next call\n",
              "  // to uploadFilesContinue from Python.\n",
              "  outputElement.steps = steps;\n",
              "\n",
              "  return _uploadFilesContinue(outputId);\n",
              "}\n",
              "\n",
              "// This is roughly an async generator (not supported in the browser yet),\n",
              "// where there are multiple asynchronous steps and the Python side is going\n",
              "// to poll for completion of each step.\n",
              "// This uses a Promise to block the python side on completion of each step,\n",
              "// then passes the result of the previous step as the input to the next step.\n",
              "function _uploadFilesContinue(outputId) {\n",
              "  const outputElement = document.getElementById(outputId);\n",
              "  const steps = outputElement.steps;\n",
              "\n",
              "  const next = steps.next(outputElement.lastPromiseValue);\n",
              "  return Promise.resolve(next.value.promise).then((value) => {\n",
              "    // Cache the last promise value to make it available to the next\n",
              "    // step of the generator.\n",
              "    outputElement.lastPromiseValue = value;\n",
              "    return next.value.response;\n",
              "  });\n",
              "}\n",
              "\n",
              "/**\n",
              " * Generator function which is called between each async step of the upload\n",
              " * process.\n",
              " * @param {string} inputId Element ID of the input file picker element.\n",
              " * @param {string} outputId Element ID of the output display.\n",
              " * @return {!Iterable<!Object>} Iterable of next steps.\n",
              " */\n",
              "function* uploadFilesStep(inputId, outputId) {\n",
              "  const inputElement = document.getElementById(inputId);\n",
              "  inputElement.disabled = false;\n",
              "\n",
              "  const outputElement = document.getElementById(outputId);\n",
              "  outputElement.innerHTML = '';\n",
              "\n",
              "  const pickedPromise = new Promise((resolve) => {\n",
              "    inputElement.addEventListener('change', (e) => {\n",
              "      resolve(e.target.files);\n",
              "    });\n",
              "  });\n",
              "\n",
              "  const cancel = document.createElement('button');\n",
              "  inputElement.parentElement.appendChild(cancel);\n",
              "  cancel.textContent = 'Cancel upload';\n",
              "  const cancelPromise = new Promise((resolve) => {\n",
              "    cancel.onclick = () => {\n",
              "      resolve(null);\n",
              "    };\n",
              "  });\n",
              "\n",
              "  // Wait for the user to pick the files.\n",
              "  const files = yield {\n",
              "    promise: Promise.race([pickedPromise, cancelPromise]),\n",
              "    response: {\n",
              "      action: 'starting',\n",
              "    }\n",
              "  };\n",
              "\n",
              "  cancel.remove();\n",
              "\n",
              "  // Disable the input element since further picks are not allowed.\n",
              "  inputElement.disabled = true;\n",
              "\n",
              "  if (!files) {\n",
              "    return {\n",
              "      response: {\n",
              "        action: 'complete',\n",
              "      }\n",
              "    };\n",
              "  }\n",
              "\n",
              "  for (const file of files) {\n",
              "    const li = document.createElement('li');\n",
              "    li.append(span(file.name, {fontWeight: 'bold'}));\n",
              "    li.append(span(\n",
              "        `(${file.type || 'n/a'}) - ${file.size} bytes, ` +\n",
              "        `last modified: ${\n",
              "            file.lastModifiedDate ? file.lastModifiedDate.toLocaleDateString() :\n",
              "                                    'n/a'} - `));\n",
              "    const percent = span('0% done');\n",
              "    li.appendChild(percent);\n",
              "\n",
              "    outputElement.appendChild(li);\n",
              "\n",
              "    const fileDataPromise = new Promise((resolve) => {\n",
              "      const reader = new FileReader();\n",
              "      reader.onload = (e) => {\n",
              "        resolve(e.target.result);\n",
              "      };\n",
              "      reader.readAsArrayBuffer(file);\n",
              "    });\n",
              "    // Wait for the data to be ready.\n",
              "    let fileData = yield {\n",
              "      promise: fileDataPromise,\n",
              "      response: {\n",
              "        action: 'continue',\n",
              "      }\n",
              "    };\n",
              "\n",
              "    // Use a chunked sending to avoid message size limits. See b/62115660.\n",
              "    let position = 0;\n",
              "    do {\n",
              "      const length = Math.min(fileData.byteLength - position, MAX_PAYLOAD_SIZE);\n",
              "      const chunk = new Uint8Array(fileData, position, length);\n",
              "      position += length;\n",
              "\n",
              "      const base64 = btoa(String.fromCharCode.apply(null, chunk));\n",
              "      yield {\n",
              "        response: {\n",
              "          action: 'append',\n",
              "          file: file.name,\n",
              "          data: base64,\n",
              "        },\n",
              "      };\n",
              "\n",
              "      let percentDone = fileData.byteLength === 0 ?\n",
              "          100 :\n",
              "          Math.round((position / fileData.byteLength) * 100);\n",
              "      percent.textContent = `${percentDone}% done`;\n",
              "\n",
              "    } while (position < fileData.byteLength);\n",
              "  }\n",
              "\n",
              "  // All done.\n",
              "  yield {\n",
              "    response: {\n",
              "      action: 'complete',\n",
              "    }\n",
              "  };\n",
              "}\n",
              "\n",
              "scope.google = scope.google || {};\n",
              "scope.google.colab = scope.google.colab || {};\n",
              "scope.google.colab._files = {\n",
              "  _uploadFiles,\n",
              "  _uploadFilesContinue,\n",
              "};\n",
              "})(self);\n",
              "</script> "
            ],
            "text/plain": [
              "<IPython.core.display.HTML object>"
            ]
          },
          "metadata": {},
          "output_type": "display_data"
        },
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Saving flink-connector-kafka-3.0.0-1.17.jar to flink-connector-kafka-3.0.0-1.17 (1).jar\n"
          ]
        }
      ],
      "source": [
        "from google.colab import files\n",
        "uploaded = files.upload()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4BkQB7QxKx0g"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "jar_filename = 'flink-connector-kafka-3.0.0-1.17.jar'\n",
        "uploaded_jar_path = os.path.abspath(next(iter(uploaded.keys())))\n",
        "os.rename(uploaded_jar_path, os.path.join(os.getcwd(), jar_filename))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6lxeVUYoLvpZ",
        "outputId": "a30a6951-a9c6-47f4-c1f2-a6c68971fc34"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Requirement already satisfied: confluent_kafka in /usr/local/lib/python3.10/dist-packages (2.1.1)\n"
          ]
        }
      ],
      "source": [
        "pip install confluent_kafka"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6n1c1RwUQzD5",
        "outputId": "afb5b20e-7b18-4af3-adc1-aff22e9b3cc8"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "['/usr/local/lib/python3.10/dist-packages/pyflink/table']\n",
            "<module 'pyflink.table.descriptors' from '/usr/local/lib/python3.10/dist-packages/pyflink/table/descriptors.py'>\n"
          ]
        }
      ],
      "source": [
        "import pyflink\n",
        "import pyflink.table\n",
        "import pyflink.table.descriptors\n",
        "print(pyflink.table.__path__)\n",
        "print(pyflink.table.descriptors)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "zqGbx6SBQ49O"
      },
      "outputs": [],
      "source": [
        "from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic\n",
        "from pyflink.table import StreamTableEnvironment, DataTypes\n",
        "from pyflink.table.descriptors import Schema, Rowtime\n",
        "from pyflink.table.udf import udf"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CmjeE9K2cncO"
      },
      "outputs": [],
      "source": [
        "from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic\n",
        "from pyflink.table import StreamTableEnvironment, DataTypes\n",
        "from pyflink.table.descriptors import Schema, Rowtime\n",
        "from pyflink.table.udf import udf\n",
        "from pyflink.table.window import Tumble\n",
        "from pyflink.table.expressions import col, lit, row_interval, concat\n",
        "from pyflink.common.serialization import SimpleStringSchema, SerializationSchema\n",
        "from pyflink.common.typeinfo import TypeInformation\n",
        "from pyflink.common import Types\n",
        "from pyflink.datastream.connectors import FlinkKafkaConsumer, FlinkKafkaProducer\n",
        "from pyflink.datastream.functions import KeyedProcessFunction, WindowFunction\n",
        "from pyflink.datastream.state import ValueStateDescriptor\n",
        "from pyflink.datastream.window import TimeWindow\n",
        "from pyflink.metrics import MetricGroup, Metric\n",
        "from pyflink.table import Row"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "FehGFu1Nd3DR"
      },
      "outputs": [],
      "source": [
        "from typing import List, Tuple, Any\n",
        "import os\n",
        "import requests\n",
        "from flask import Flask, request, jsonify"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "qVQWz8DDVaRJ",
        "outputId": "4e99bb52-96d0-4a00-d194-e728700e1bd0"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Requirement already satisfied: kafka-python in /usr/local/lib/python3.10/dist-packages (2.0.2)\n"
          ]
        }
      ],
      "source": [
        "pip install kafka-python"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "SIFkBd2rNzM5",
        "outputId": "21747916-fea8-4131-aa2e-9b1b15c4b461"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Requirement already satisfied: pymongo[srv] in /usr/local/lib/python3.10/dist-packages (3.13.0)\n",
            "Collecting dnspython<3.0.0,>=1.16.0 (from pymongo[srv])\n",
            "  Downloading dnspython-2.3.0-py3-none-any.whl (283 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m283.7/283.7 kB\u001b[0m \u001b[31m6.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hInstalling collected packages: dnspython\n",
            "Successfully installed dnspython-2.3.0\n"
          ]
        }
      ],
      "source": [
        "!pip install pymongo[srv]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4-EYHTzEfc8n"
      },
      "outputs": [],
      "source": [
        "from pymongo import MongoClient\n",
        "import json\n",
        "import random\n",
        "import time\n",
        "from confluent_kafka import Producer, Consumer, KafkaError\n",
        "import json\n",
        "import random\n",
        "import time\n",
        "from confluent_kafka import Producer, Consumer, KafkaError"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "background_save": true,
          "base_uri": "https://localhost:8080/",
          "height": 132
        },
        "id": "-vlv65W7hBd-",
        "outputId": "175aa42b-edfe-4559-c7d0-e7ebe2fcf581"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "\u001b[33mWARNING: Skipping kafka as it is not installed.\u001b[0m\u001b[33m\n",
            "\u001b[0mFound existing installation: kafka-python 2.0.2\n",
            "Uninstalling kafka-python-2.0.2:\n",
            "  Would remove:\n",
            "    /usr/local/lib/python3.10/dist-packages/kafka/*\n",
            "    /usr/local/lib/python3.10/dist-packages/kafka_python-2.0.2.dist-info/*\n",
            "Proceed (Y/n)? y\n",
            "  Successfully uninstalled kafka-python-2.0.2\n",
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting kafka-python\n",
            "  Using cached kafka_python-2.0.2-py2.py3-none-any.whl (246 kB)\n",
            "Installing collected packages: kafka-python\n",
            "Successfully installed kafka-python-2.0.2\n"
          ]
        }
      ],
      "source": [
        "pip uninstall kafka\n",
        "pip uninstall kafka-python\n",
        "pip install kafka-python"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "#from pyflink.table.descriptors import Schema, Kafka, Json, Rowtime, FileSystem\n",
        "#from pyflink.table.expressions import col, lit, row_interval, concat, substring"
      ],
      "metadata": {
        "id": "sZlGf9aapCQp"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "# Create execution environment\n",
        "env = StreamExecutionEnvironment.get_execution_environment()\n",
        "env.set_stream_time_characteristic(TimeCharacteristic.EventTime)\n",
        "\n",
        "# Create table environment\n",
        "t_env = StreamTableEnvironment.create(env)"
      ],
      "metadata": {
        "id": "XiP0Lm6wqvV9"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "from pymongo.mongo_client import MongoClient\n",
        "from pymongo.server_api import ServerApi\n",
        "\n",
        "uri = \"mongodb+srv://scrapper:shaylin1396*M@cluster0.cco0jwv.mongodb.net/?retryWrites=true&w=majority\"\n",
        "\n",
        "# Create a new client and connect to the server\n",
        "client = MongoClient(uri, server_api=ServerApi('1'))\n",
        "\n",
        "# Send a ping to confirm a successful connection\n",
        "try:\n",
        "    client.admin.command('ping')\n",
        "    print(\"Pinged your deployment. You successfully connected to MongoDB!\")\n",
        "except Exception as e:\n",
        "    print(e)"
      ],
      "metadata": {
        "id": "c-vuydtUrtgx",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "1fa43f49-834e-41a6-f4f7-541bda6f3ae4"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Pinged your deployment. You successfully connected to MongoDB!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Set the configuration properties\n",
        "kafka_config = {\n",
        "    \"bootstrap.servers\": \"pkc-6ojv2.us-west4.gcp.confluent.cloud:9092\",\n",
        "    \"security.protocol\": \"SASL_SSL\",\n",
        "    \"sasl.mechanisms\": \"PLAIN\",\n",
        "    \"sasl.username\": \"RFU2BCEDGB2UJRWD\",\n",
        "    \"sasl.password\": \"rjOrdR009uQUx8F67ObMn6on0IAgEtSunMFIKR2KK17SVJqQ1gh+9CmTT9NXiAjp\",\n",
        "    \"group.id\": \"my_group.Id\"}"
      ],
      "metadata": {
        "id": "nMjL7f1Jtdtc"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "# Define Kafka topic and schema\n",
        "kafka_topic = 'KafkaTopicIN'\n",
        "kafka_schema = DataTypes.ROW([\n",
        "    DataTypes.FIELD('product_url', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('brand', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('title', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('price_hight', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('price_low', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('sku', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('availability', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('store_id', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('category_id', DataTypes.STRING()),\n",
        "    DataTypes.FIELD('date', DataTypes.STRING())\n",
        "])"
      ],
      "metadata": {
        "id": "pRMih_5Yvji4"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyflink.table import DataTypes\n",
        "from pyflink.table.descriptors import Schema, Kafka, Json, Rowtime\n",
        "from pyflink.table.catalog import FileSystem"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 600
        },
        "id": "XMdyN-4w2jwf",
        "outputId": "bc83cfae-e2e8-4c20-cd0d-6f8775e4fda7"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "error",
          "ename": "ImportError",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
            "\u001b[0;31mImportError\u001b[0m                               Traceback (most recent call last)",
            "\u001b[0;32m<ipython-input-1-203af9e2c559>\u001b[0m in \u001b[0;36m<cell line: 2>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mDataTypes\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdescriptors\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mSchema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mKafka\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mJson\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mRowtime\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      3\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcatalog\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mFileSystem\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mImportError\u001b[0m: cannot import name 'Kafka' from 'pyflink.table.descriptors' (/usr/local/lib/python3.10/dist-packages/pyflink/table/descriptors.py)",
            "",
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0;32m\nNOTE: If your import is failing due to a missing package, you can\nmanually install dependencies using either !pip or !apt.\n\nTo view examples of installing some common dependencies, click the\n\"Open Examples\" button below.\n\u001b[0;31m---------------------------------------------------------------------------\u001b[0m\n"
          ],
          "errorDetails": {
            "actions": [
              {
                "action": "open_url",
                "actionText": "Open Examples",
                "url": "/notebooks/snippets/importing_libraries.ipynb"
              }
            ]
          }
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pykafka"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "NX4qspTm26qb",
        "outputId": "c56c6922-ec3a-4b5c-be80-9475fe39ffc0"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting pykafka\n",
            "  Downloading pykafka-2.8.0.tar.gz (141 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m141.6/141.6 kB\u001b[0m \u001b[31m3.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.10/dist-packages (from pykafka) (1.16.0)\n",
            "Collecting kazoo==2.5.0 (from pykafka)\n",
            "  Downloading kazoo-2.5.0-py2.py3-none-any.whl (129 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m129.4/129.4 kB\u001b[0m \u001b[31m13.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: tabulate in /usr/local/lib/python3.10/dist-packages (from pykafka) (0.8.10)\n",
            "Building wheels for collected packages: pykafka\n",
            "  Building wheel for pykafka (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pykafka: filename=pykafka-2.8.0-py2.py3-none-any.whl size=171508 sha256=addf881d63b4d672284885b97838f3282730eeba2e813a30541a624b5d749836\n",
            "  Stored in directory: /root/.cache/pip/wheels/07/bf/52/308d7d0eab446c3ddeb5e02795181c68a8c4b3d8921914acd9\n",
            "Successfully built pykafka\n",
            "Installing collected packages: kazoo, pykafka\n",
            "Successfully installed kazoo-2.5.0 pykafka-2.8.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "t_env.connect(Kafka()\n",
        "              .version('universal')\n",
        "              .topic(KAFKA_TOPIC)\n",
        "              .start_from_earliest()\n",
        "              .properties(kafka_props)\n",
        "              .deserialization_schema(Json()\n",
        "                                       .fail_on_missing_field(False)\n",
        "                                       .ignore_parse_errors()\n",
        "                                       .schema(DataTypes.ROW([\n",
        "\n",
        "DataTypes.FIELD('product_url', DataTypes.STRING()),\n",
        "                                           DataTypes.FIELD('brand',\n",
        "DataTypes.STRING()),\n",
        "                                           DataTypes.FIELD('title',\n",
        "DataTypes.STRING()),\n",
        "\n",
        "DataTypes.FIELD('price_hight', DataTypes.STRING()),\n",
        "\n",
        "DataTypes.FIELD('price_low', DataTypes.STRING()),\n",
        "                                           DataTypes.FIELD('sku',\n",
        "DataTypes.STRING()),\n",
        "\n",
        "DataTypes.FIELD('availability', DataTypes.STRING()),\n",
        "                                           DataTypes.FIELD('store_id',\n",
        "DataTypes.STRING()),\n",
        "\n",
        "DataTypes.FIELD('category_id', DataTypes.STRING()),\n",
        "                                           DataTypes.FIELD('date',\n",
        "DataTypes.STRING())\n",
        "                                       ]))\n",
        "              ) \\\n",
        "    .with_format(Json()\n",
        "                 .fail_on_missing_field(False)\n",
        "                 .ignore_parse_errors()\n",
        "                 .schema(DataTypes.ROW([\n",
        "                     DataTypes.FIELD('product_url', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('brand', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('title', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('price_hight', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('price_low', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('sku', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('availability', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('store_id', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('category_id', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('date', DataTypes.STRING())\n",
        "                 ]))\n",
        "                 .for_produced_schema(DataTypes.ROW([\n",
        "                     DataTypes.FIELD('product_url', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('brand', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('title', DataTypes.STRING()),\n",
        "                     DataTypes.FIELD('price_hight', DataTypes\n",
        "\n",
        "# define a Flink job to perform the \"good deal\" analysis\n",
        "\n",
        "\n",
        "job_def = {\n",
        "    \"entryClass\": \"com.example.GoodDealJob\",\n",
        "    \"programArgs\": [\n",
        "        \"--product_id\", \"\",\n",
        "        \"--timestamp\", \"\",\n",
        "        \"--price_high\", \"\"\n",
        "    ]\n",
        "}\n",
        "\n",
        "@app.route('/good_deal', methods=['GET'])\n",
        "def good_deal():\n",
        "    # parse request parameters\n",
        "    product_id = request.args.get('product_id')\n",
        "    timestamp = request.args.get('timestamp')\n",
        "    price_high = request.args.get('price_high')\n",
        "\n",
        "    # submit the Flink job\n",
        "    job_def['programArgs'][1] = product_id\n",
        "    job_def['programArgs'][3] = timestamp\n",
        "    job_def['programArgs'][5] = price_high\n",
        "    response = requests.post(os.environ['FLINK_ENDPOINT'] +\n",
        "'/jars/upload', files={'jarfile': open('good-deal-job.jar', 'rb')})\n",
        "    jar_id = response.json()['filename'].replace('.jar', '')\n",
        "    response = requests.post(os.environ['FLINK_ENDPOINT'] + '/jobs/' +\n",
        "jar_id + '/run', json=job_def)\n",
        "\n",
        "    # wait for the job to finish\n",
        "    job_id = response.json()['jobid']\n",
        "    response = requests.get(os.environ['FLINK_ENDPOINT'] + '/jobs/' + job_id)\n",
        "    while response.json()['state'] != 'FINISHED':\n",
        "        response = requests.get(os.environ['FLINK_ENDPOINT'] +\n",
        "'/jobs/' + job_id)\n",
        "\n",
        "    # get the result from the MongoDB collection\n",
        "    good_deals = db.good_deals\n",
        "    result = good_deals.find_one({'product_id': product_id,\n",
        "'timestamp': timestamp})\n",
        "    good_deal = result['good_deal']\n",
        "    return jsonify({'product_id': product_id, 'timestamp': timestamp,\n",
        "'good_deal': good_deal})"
      ],
      "metadata": {
        "id": "IOkirRdBBBSr",
        "outputId": "e907fa6e-61d0-4f38-d3ce-25957af2b060",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 131
        }
      },
      "execution_count": 5,
      "outputs": [
        {
          "output_type": "error",
          "ename": "SyntaxError",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-5-503b78bb8354>\"\u001b[0;36m, line \u001b[0;32m51\u001b[0m\n\u001b[0;31m    DataTypes.FIELD('price_hight', DataTypes\u001b[0m\n\u001b[0m                   ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m '(' was never closed\n"
          ]
        }
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": [],
      "name": "kafkaconsumer1.ipynb",
      "authorship_tag": "ABX9TyP2jbEsxDC9bljDyxbsa4+F",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}