kafkaconsumerspark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyMULA/qVMo+0Bgx2NJOvAHW",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/aomidvar/alibabainpersia/blob/master/kafkaconsumerspark.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": {
        "id": "1tbWQXpfbOT_"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "import sys\n",
        "import urllib\n",
        "\n",
        "# Define the Spark version to install\n",
        "spark_version = \"3.2.0\"\n",
        "\n",
        "# Download Spark\n",
        "filename = f\"spark-{spark_version}-bin-hadoop3.2.tgz\"\n",
        "url = f\"https://downloads.apache.org/spark/spark-3.2.4/spark-3.2.4-bin-hadoop2.7.tgz\"\n",
        "urllib.request.urlretrieve(url, filename)\n",
        "\n",
        "# Extract Spark\n",
        "!tar -xvf $filename\n"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Set up environment variables\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-1.11.0-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = f\"/content/spark-3.2.4-bin-hadoop2.7\"\n",
        "\n",
        "# Update PATH environment variable\n",
        "os.environ[\"PATH\"] += \":\" + os.environ[\"SPARK_HOME\"] + \"/bin:\" + os.environ[\"SPARK_HOME\"] + \"/sbin\"\n",
        "\n",
        "# Install findspark library to help locate Spark\n",
        "!pip install findspark\n"
      ],
      "metadata": {
        "id": "vm-iYVdZdwCU"
      },
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import os"
      ],
      "metadata": {
        "id": "CITbaydZfTNy"
      },
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install -U pyarrow"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "KyEY_N0wer5v",
        "outputId": "0010ab02-c7c9-4a98-df7b-4afc17ec683c"
      },
      "execution_count": 3,
      "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: pyarrow in /usr/local/lib/python3.10/dist-packages (9.0.0)\n",
            "Collecting pyarrow\n",
            "  Downloading pyarrow-12.0.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (38.9 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m38.9/38.9 MB\u001b[0m \u001b[31m35.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: numpy>=1.16.6 in /usr/local/lib/python3.10/dist-packages (from pyarrow) (1.22.4)\n",
            "Installing collected packages: pyarrow\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",
            "\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",
            "pandas-gbq 0.17.9 requires pyarrow<10.0dev,>=3.0.0, but you have pyarrow 12.0.0 which is incompatible.\u001b[0m\u001b[31m\n",
            "\u001b[0mSuccessfully installed pyarrow-12.0.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pyspark[sql-kafka]\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 346
        },
        "id": "LhUsSgwQe7ki",
        "outputId": "187b6464-baa7-424f-ded2-49766de8fe37"
      },
      "execution_count": 15,
      "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 pyspark[sql-kafka]\n",
            "  Downloading pyspark-3.4.0.tar.gz (310.8 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m310.8/310.8 MB\u001b[0m \u001b[31m2.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "\u001b[33mWARNING: pyspark 3.4.0 does not provide the extra 'sql-kafka'\u001b[0m\u001b[33m\n",
            "\u001b[0mRequirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark[sql-kafka]) (0.10.9.7)\n",
            "Building wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.4.0-py2.py3-none-any.whl size=311317130 sha256=c2d7910fc2cc75caeb77dc132982dabb8ef66beedd903bb37afee51b83c2e226\n",
            "  Stored in directory: /root/.cache/pip/wheels/7b/1b/4b/3363a1d04368e7ff0d408e57ff57966fcdf00583774e761327\n",
            "Successfully built pyspark\n",
            "Installing collected packages: pyspark\n",
            "Successfully installed pyspark-3.4.0\n"
          ]
        },
        {
          "output_type": "display_data",
          "data": {
            "application/vnd.colab-display-data+json": {
              "pip_warning": {
                "packages": [
                  "pyspark"
                ]
              }
            }
          },
          "metadata": {}
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import findspark\n",
        "findspark.init()\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "\n",
        "spark = SparkSession.builder \\\n",
        "    .appName(\"SparkColabIntegration\") \\\n",
        "    .getOrCreate()\n"
      ],
      "metadata": {
        "id": "8_y7fM3Id5Sx"
      },
      "execution_count": 8,
      "outputs": []
    },
    {
      "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": "PurA-pTMhMIu"
      },
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pymongo[srv]"
      ],
      "metadata": {
        "id": "ni0BManYhQTb",
        "outputId": "c8ceca1e-d123-4960-f520-cde3548d0a71",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "execution_count": 11,
      "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 pymongo[srv]\n",
            "  Downloading pymongo-4.3.3-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (492 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m492.9/492.9 kB\u001b[0m \u001b[31m10.2 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting 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[31m22.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hInstalling collected packages: dnspython, pymongo\n",
            "Successfully installed dnspython-2.3.0 pymongo-4.3.3\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "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": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "9iXKgTeMeayp",
        "outputId": "3b850605-ff9a-439d-ec2d-caf856e19837"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Pinged your deployment. You successfully connected to MongoDB!\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Install the necessary dependencies\n",
        "!pip install confluent-kafka\n",
        "!pip install confluent-kafka[avro]"
      ],
      "metadata": {
        "id": "gy6x8oAHkpg4",
        "outputId": "9d4bb131-7ec5-4c10-d5ca-ec3d63f7b13c",
        "colab": {
          "base_uri": "https://localhost:8080/"
        }
      },
      "execution_count": 17,
      "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[31m35.7 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",
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Requirement already satisfied: confluent-kafka[avro] in /usr/local/lib/python3.10/dist-packages (2.1.1)\n",
            "Collecting avro<2,>=1.11.1 (from confluent-kafka[avro])\n",
            "  Downloading avro-1.11.1.tar.gz (84 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m84.2/84.2 kB\u001b[0m \u001b[31m3.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Installing build dependencies ... \u001b[?25l\u001b[?25hdone\n",
            "  Getting requirements to build wheel ... \u001b[?25l\u001b[?25hdone\n",
            "  Preparing metadata (pyproject.toml) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: requests in /usr/local/lib/python3.10/dist-packages (from confluent-kafka[avro]) (2.27.1)\n",
            "Collecting fastavro>=1.0 (from confluent-kafka[avro])\n",
            "  Downloading fastavro-1.7.4-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[31m27.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: urllib3<1.27,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests->confluent-kafka[avro]) (1.26.15)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests->confluent-kafka[avro]) (2022.12.7)\n",
            "Requirement already satisfied: charset-normalizer~=2.0.0 in /usr/local/lib/python3.10/dist-packages (from requests->confluent-kafka[avro]) (2.0.12)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests->confluent-kafka[avro]) (3.4)\n",
            "Building wheels for collected packages: avro\n",
            "  Building wheel for avro (pyproject.toml) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for avro: filename=avro-1.11.1-py2.py3-none-any.whl size=116718 sha256=a1767186da9a3545fe7fb1fdef6c3b27213aedbf90a55ec578e4b737706ff659\n",
            "  Stored in directory: /root/.cache/pip/wheels/7c/a2/23/0ac3a569e5283890cef1fc901e9a5098df752ce2488c911720\n",
            "Successfully built avro\n",
            "Installing collected packages: fastavro, avro\n",
            "Successfully installed avro-1.11.1 fastavro-1.7.4\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql import functions as F\n",
        "from pyspark.sql.types import StructType\n",
        "from pyspark.sql.types import StringType, StructField, StructType\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import from_json\n",
        "from pyspark.sql.types import StringType, StructField, StructType\n",
        "\n",
        "# Set up Spark session\n",
        "spark = SparkSession.builder \\\n",
        "    .appName(\"KafkaConsumer\") \\\n",
        "    .getOrCreate()\n",
        "\n",
        "# Define Kafka topic and schema\n",
        "kafka_topic = \"KafkaTopicIN\"\n",
        "kafka_schema = StructType([\n",
        "    StructField(\"product_url\", StringType()),\n",
        "    StructField(\"brand\", StringType()),\n",
        "    StructField(\"title\", StringType()),\n",
        "    StructField(\"price_hight\", StringType()),\n",
        "    StructField(\"price_low\", StringType()),\n",
        "    StructField(\"sku\", StringType()),\n",
        "    StructField(\"availability\", StringType()),\n",
        "    StructField(\"store_id\", StringType()),\n",
        "    StructField(\"category_id\", StringType()),\n",
        "    StructField(\"date\", StringType())\n",
        "])\n",
        "\n",
        "# Set Kafka configuration\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\": \"your_group_id\"\n",
        "}\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "DcnZ9LtqiWZC"
      },
      "execution_count": 20,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "\n"
      ],
      "metadata": {
        "id": "zHZnA0qVmMqX"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql import SparkSession\n",
        "\n",
        "\n",
        "\n",
        "# Define Kafka topic and schema\n",
        "kafka_topic = \"KafkaTopicIN\"\n",
        "\n",
        "# Set Kafka configuration\n",
        "kafka_config = {\n",
        "    \"kafka.bootstrap.servers\": \"pkc-6ojv2.us-west4.gcp.confluent.cloud:9092\",\n",
        "    \"kafka.security.protocol\": \"SASL_SSL\",\n",
        "    \"kafka.sasl.mechanisms\": \"PLAIN\",\n",
        "    \"kafka.sasl.username\": \"RFU2BCEDGB2UJRWD\",\n",
        "    \"kafka.sasl.password\": \"rjOrdR009uQUx8F67ObMn6on0IAgEtSunMFIKR2KK17SVJqQ1gh+9CmTT9NXiAjp\",\n",
        "    \"kafka.group.id\": \"your_group_id\"\n",
        "}\n",
        "\n",
        "# Read from Kafka topic\n",
        "df = spark \\\n",
        "    .readStream \\\n",
        "    .format(\"kafka\") \\\n",
        "    .option(\"kafka.bootstrap.servers\", kafka_config[\"kafka.bootstrap.servers\"]) \\\n",
        "    .option(\"kafka.security.protocol\", kafka_config[\"kafka.security.protocol\"]) \\\n",
        "    .option(\"kafka.sasl.mechanism\", kafka_config[\"kafka.sasl.mechanisms\"]) \\\n",
        "    .option(\"kafka.sasl.username\", kafka_config[\"kafka.sasl.username\"]) \\\n",
        "\n"
      ],
      "metadata": {
        "id": "9Qw9KC28rJSz"
      },
      "execution_count": 27,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "ADvVndyWrsuZ"
      },
      "execution_count": 27,
      "outputs": []
    }
  ]
}