pyspark-kafka.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "23/12/16 19:35:53 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "23/12/16 19:35:54 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n",
      "23/12/16 19:35:54 WARN Utils: Service 'SparkUI' could not bind on port 4041. Attempting port 4042.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "23/12/16 19:36:04 WARN GarbageCollectionMetrics: To enable non-built-in garbage collector(s) List(G1 Concurrent GC), users should configure it(them) to spark.eventLog.gcMetrics.youngGenerationGarbageCollectors or spark.eventLog.gcMetrics.oldGenerationGarbageCollectors\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession\\\n",
    "    .builder\\\n",
    "    .master(\"local[2]\")\\\n",
    "    .appName('pyspark-kafka')\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "Failed to find data source: kafka. Please deploy the application as per the deployment section of Structured Streaming + Kafka Integration Guide.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb Cell 3\u001b[0m line \u001b[0;36m6\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=0'>1</a>\u001b[0m df \u001b[39m=\u001b[39m spark\u001b[39m.\u001b[39mreadStream\\\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=1'>2</a>\u001b[0m         \u001b[39m.\u001b[39mformat(\u001b[39m\"\u001b[39m\u001b[39mkafka\u001b[39m\u001b[39m\"\u001b[39m)\\\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=2'>3</a>\u001b[0m         \u001b[39m.\u001b[39moption(\u001b[39m\"\u001b[39m\u001b[39mkafka.bootstrap.servers\u001b[39m\u001b[39m\"\u001b[39m, \u001b[39m\"\u001b[39m\u001b[39m192.168.1.100:9092\u001b[39m\u001b[39m\"\u001b[39m)\\\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=3'>4</a>\u001b[0m         \u001b[39m.\u001b[39moption(\u001b[39m\"\u001b[39m\u001b[39msubscribe\u001b[39m\u001b[39m\"\u001b[39m, \u001b[39m\"\u001b[39m\u001b[39mjson_topic\u001b[39m\u001b[39m\"\u001b[39m)\\\n\u001b[1;32m      <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=4'>5</a>\u001b[0m         \u001b[39m.\u001b[39moption(\u001b[39m\"\u001b[39m\u001b[39mstartingOffsets\u001b[39m\u001b[39m\"\u001b[39m, \u001b[39m\"\u001b[39m\u001b[39mearliest\u001b[39m\u001b[39m\"\u001b[39m)\\\n\u001b[0;32m----> <a href='vscode-notebook-cell:/Users/biryani/Documents/Python/misc-prj/pyspark-kafka.ipynb#W2sZmlsZQ%3D%3D?line=5'>6</a>\u001b[0m         \u001b[39m.\u001b[39mload()\n",
      "File \u001b[0;32m~/anaconda3/lib/python3.11/site-packages/pyspark/sql/streaming/readwriter.py:304\u001b[0m, in \u001b[0;36mDataStreamReader.load\u001b[0;34m(self, path, format, schema, **options)\u001b[0m\n\u001b[1;32m    302\u001b[0m     \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_df(\u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_jreader\u001b[39m.\u001b[39mload(path))\n\u001b[1;32m    303\u001b[0m \u001b[39melse\u001b[39;00m:\n\u001b[0;32m--> 304\u001b[0m     \u001b[39mreturn\u001b[39;00m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_df(\u001b[39mself\u001b[39m\u001b[39m.\u001b[39m_jreader\u001b[39m.\u001b[39mload())\n",
      "File \u001b[0;32m~/anaconda3/lib/python3.11/site-packages/py4j/java_gateway.py:1322\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1316\u001b[0m command \u001b[39m=\u001b[39m proto\u001b[39m.\u001b[39mCALL_COMMAND_NAME \u001b[39m+\u001b[39m\\\n\u001b[1;32m   1317\u001b[0m     \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mcommand_header \u001b[39m+\u001b[39m\\\n\u001b[1;32m   1318\u001b[0m     args_command \u001b[39m+\u001b[39m\\\n\u001b[1;32m   1319\u001b[0m     proto\u001b[39m.\u001b[39mEND_COMMAND_PART\n\u001b[1;32m   1321\u001b[0m answer \u001b[39m=\u001b[39m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mgateway_client\u001b[39m.\u001b[39msend_command(command)\n\u001b[0;32m-> 1322\u001b[0m return_value \u001b[39m=\u001b[39m get_return_value(\n\u001b[1;32m   1323\u001b[0m     answer, \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mgateway_client, \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mtarget_id, \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mname)\n\u001b[1;32m   1325\u001b[0m \u001b[39mfor\u001b[39;00m temp_arg \u001b[39min\u001b[39;00m temp_args:\n\u001b[1;32m   1326\u001b[0m     \u001b[39mif\u001b[39;00m \u001b[39mhasattr\u001b[39m(temp_arg, \u001b[39m\"\u001b[39m\u001b[39m_detach\u001b[39m\u001b[39m\"\u001b[39m):\n",
      "File \u001b[0;32m~/anaconda3/lib/python3.11/site-packages/pyspark/errors/exceptions/captured.py:185\u001b[0m, in \u001b[0;36mcapture_sql_exception.<locals>.deco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    181\u001b[0m converted \u001b[39m=\u001b[39m convert_exception(e\u001b[39m.\u001b[39mjava_exception)\n\u001b[1;32m    182\u001b[0m \u001b[39mif\u001b[39;00m \u001b[39mnot\u001b[39;00m \u001b[39misinstance\u001b[39m(converted, UnknownException):\n\u001b[1;32m    183\u001b[0m     \u001b[39m# Hide where the exception came from that shows a non-Pythonic\u001b[39;00m\n\u001b[1;32m    184\u001b[0m     \u001b[39m# JVM exception message.\u001b[39;00m\n\u001b[0;32m--> 185\u001b[0m     \u001b[39mraise\u001b[39;00m converted \u001b[39mfrom\u001b[39;00m \u001b[39mNone\u001b[39;00m\n\u001b[1;32m    186\u001b[0m \u001b[39melse\u001b[39;00m:\n\u001b[1;32m    187\u001b[0m     \u001b[39mraise\u001b[39;00m\n",
      "\u001b[0;31mAnalysisException\u001b[0m: Failed to find data source: kafka. Please deploy the application as per the deployment section of Structured Streaming + Kafka Integration Guide."
     ]
    }
   ],
   "source": [
    "# df = spark.readStream\\\n",
    "#         .format(\"kafka\")\\\n",
    "#         .option(\"kafka.bootstrap.servers\", \"192.168.1.100:9092\")\\\n",
    "#         .option(\"subscribe\", \"json_topic\")\\\n",
    "#         .option(\"startingOffsets\", \"earliest\")\\\n",
    "#         .load()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}