doc.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9fb8a47c",
   "metadata": {},
   "source": [
    "### <center>QuakeMatch </center>\n",
    "### <center>Luca Longo, Università degli Studi di Catania </center>\n",
    "<img src=\"https://user-images.githubusercontent.com/50534107/256270970-f0de1c54-ba7f-46a0-ab32-c4ad00e6b26e.svg\" alt=\"QuakeMatch\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5fc9b2a",
   "metadata": {},
   "source": [
    "## Table of contents:\n",
    "* [Introduction](#Introduction)\n",
    "* [Quake Match Pipeline](#QuakeMatchPipeline)\n",
    "* [Data Ingestion](#dataIngestion)\n",
    "    * [Python Script](#pythonScript)\n",
    "    * [Logstash](#logstash)\n",
    "* [Streaming with Kafka](#kafka)\n",
    "* [Processing with Apache Spark](#spark)\n",
    "* [Data indexing with Elasticsearch](#elasticsearch)\n",
    "* [Data visualization with Kibana](#kibana)\n",
    "* [Requirements](#requirements)\n",
    "* [Usage](#usage)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "360c3c59",
   "metadata": {},
   "source": [
    "# Introduction <a class=\"anchor\" id=\"Introduction\"></a>\n",
    "\n",
    "<p>\n",
    "QuakeMatch is a tool for matching detection of antipodal earthquakes that use sismics data to look for a match between two or more events. It uses technologies such as Logstash, Kafka with Kafka Zookeeper, Apache Spark, Elasticsearch and Kibana for data ingestion, filtering, elaboration and results visualization.\n",
    "</p>\n",
    "\n",
    "## Quake Match Pipeline<a class=\"anchor\" id=\"QuakeMatchPipeline\"></a>\n",
    "<p>\n",
    "    <img src=\"https://user-images.githubusercontent.com/50534107/256268865-21fcd7c9-1762-45da-8a8e-aed7e15b7468.png\" alt=\"Pipeline\" />\n",
    "</p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "39814127",
   "metadata": {},
   "source": [
    "# Data Ingestion <a class=\"anchor\" id=\"dataIngestion\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0566e125-8383-4253-b1de-4aeaa54de263",
   "metadata": {},
   "source": [
    "## Python Script <a class=\"anchor\" id=\"pythonScript\"></a>\n",
    "The following code allows you to generate a file containing all the urls needed for the API requests made by Logstash:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d65b23ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "\n",
    "base_url = \"https://www.seismicportal.eu/fdsnws/event/1/query?limit=7000&start={}&end={}\"\n",
    "\n",
    "start_date = datetime.datetime(1998, 7, 19)\n",
    "end_date = datetime.datetime(2023, 7, 19)\n",
    "\n",
    "current_date = start_date\n",
    "links = []\n",
    "\n",
    "i=0\n",
    "while current_date <= end_date:\n",
    "    start_time = current_date.strftime(\"%Y-%m-%dT%H:%M:%S.0\")\n",
    "    end_time = (current_date + datetime.timedelta(days=1) - datetime.timedelta(seconds=1)).strftime(\"%Y-%m-%dT%H:%M:%S.0\")\n",
    "    link = base_url.format(start_time, end_time)\n",
    "    links.append(link)\n",
    "    current_date += datetime.timedelta(days=1)\n",
    "    i=i+1\n",
    "\n",
    "with open(\"./logstash/seismic_portal_links.txt\", \"w\") as file:\n",
    "    for link in links:\n",
    "        file.write(link + \"\\n\")\n",
    "\n",
    "\n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3041328e",
   "metadata": {},
   "source": [
    "You can choose the date range by modifying the start_date and end_date values"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1d61b02-4b74-4170-a902-a4c96e675265",
   "metadata": {},
   "source": [
    "## Logstash <a class=\"anchor\" id=\"logstash\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d695379",
   "metadata": {},
   "source": [
    "Logstash will read the file generated by the python script, execute the API requests to Seismic Portal, filter the data and send the generated messages to Kafka."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dad12a7b",
   "metadata": {},
   "source": [
    "# Streaming with Kafka <a class=\"anchor\" id=\"kafka\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b7d3fbfb",
   "metadata": {},
   "source": [
    "Messages sent from Logstash to Kafka will be in the \"earthquakes\" topic. Each message will present five lists including all timestamps, regions, latitudes, longitudes and magnitudes of all events"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1a5f3920",
   "metadata": {},
   "source": [
    "# Processing with Apache Spark <a class=\"anchor\" id=\"spark\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c952b91",
   "metadata": {},
   "source": [
    "Spark will take care of the following tasks:\n",
    "<ul>\n",
    "    <li>Taking messages from the Kafka topic \"earthquakes\"</li>\n",
    "    <li>Filtering by magnitude ≥ 5.5</li>\n",
    "    <li>Antipode calculation</li>\n",
    "    <li>Retrieving a match with one or more events whose latitude and longitude is in the range of ± 30° with respect to the calculated antipode and which occurred within the following three days</li>\n",
    "    <li>Creating a CSV file with the obtained data</li>\n",
    "    <li>Creating an index on Elasticsearch</li>\n",
    "    <li>Data indexing</li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54966044",
   "metadata": {},
   "source": [
    "# Data indexing with Elasticsearch <a class=\"anchor\" id=\"elasticsearch\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0580de8e",
   "metadata": {},
   "source": [
    "The data will be indexed in Elasticsearch under the index \"earthquakes\" and mapping \"earthquake_mapping\". Each item will have the following fields:\n",
    "<ul>\n",
    "    <li>unique id</li>\n",
    "    <li>timestamp</li>\n",
    "    <li>region</li>\n",
    "    <li>latitude</li>\n",
    "    <li>longitude</li>\n",
    "    <li>latitude_antipode</li>\n",
    "    <li>longitude_antipode</li>\n",
    "    <li>matches</li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d7eca74",
   "metadata": {},
   "source": [
    "# Data visualization with Kibana <a class=\"anchor\" id=\"kibana\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ebaf6ca",
   "metadata": {},
   "source": [
    "<img src=\"https://user-images.githubusercontent.com/50534107/256273468-eec68964-ff00-4630-9f45-3998e94b6037.png\" alt=\"kibana\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a3682e6f",
   "metadata": {},
   "source": [
    "The Kibana dashboard will show five lenses: the first indicates the number of matches that QuakeMatch was able to detect, followed by a descriptive table of the matches and finally three other lenses that show the data of the earthquakes occurred in Italy, the distribution of the number of earthquakes by magnitude and the distribution of earthquakes in the european area."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2295430a",
   "metadata": {},
   "source": [
    "# Requirements <a class=\"anchor\" id=\"requirements\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b05d42c",
   "metadata": {},
   "source": [
    "<ul>\n",
    "    <li>Docker</li>\n",
    "    <li>Python</li>\n",
    "    <li>wget</li>\n",
    "    <li>A solution with, at least, 16GB of RAM</li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1087895",
   "metadata": {},
   "source": [
    "# Usage <a class=\"anchor\" id=\"usage\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77f8cbb0-af8c-4f90-93e2-fd58f34978cb",
   "metadata": {},
   "source": [
    "<li>Install Docker in your system, than run the following command for create a docker network:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "d39ea3dd-0d75-49f9-92ae-f81830bc3eac",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "d3da4d4f72bd50aa53c5827e7eff7bffa127e1ba376254cc650da027198a93a4\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker network create kafka-network"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3ea528b5-6e54-44fb-bba6-a1aa40e08f29",
   "metadata": {},
   "source": [
    "<li>Create a container with Kafka Zookeeper:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "f0cc9e9c-efac-4455-a5a9-603bf0fd4e11",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7d3c4f34e5e45df23744b665c41a4d849ab3c4ec8871044e14647f0d671b8c37\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d \\\n",
    "  --name zookeeper \\\n",
    "  --network kafka-network \\\n",
    "  -p 2181:2181 \\\n",
    "  -e ZOOKEEPER_CLIENT_PORT=2181 \\\n",
    "  confluentinc/cp-zookeeper:latest"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85fa14bb-68f2-4b09-ab89-54ca167d96c7",
   "metadata": {},
   "source": [
    "<li>Wait Zookeper to be fully loaded, then create a container with Kafka:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "6dbbb62e-74bb-4e6e-a744-39677b39ed0d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "98bcbbe2498c184d3c7a413b4fcac6ee27b376e4ed1277b463a3843ff24688c1\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d \\\n",
    "  --name kafka \\\n",
    "  --network kafka-network \\\n",
    "  -p 9092:9092 \\\n",
    "  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \\\n",
    "  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \\\n",
    "  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \\\n",
    "  confluentinc/cp-kafka:latest"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9bf63ac-c5f3-47a5-8943-951b568cebde",
   "metadata": {},
   "source": [
    "<li>And a container with Kafka UI:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ea33a13c-81aa-463d-8a18-3db76c7abf93",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2efa12b6727e5a25f0b34b71b56f22c5c859d142a1d8af8ee1a14e56f197b8ca\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d \\\n",
    "  --name kafka-ui \\\n",
    "  --network kafka-network \\\n",
    "  -p 8080:8080 \\\n",
    "  -e KAFKA_CLUSTERS_0_NAME=local \\\n",
    "  -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092 \\\n",
    "  -e KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181 \\\n",
    "  -e KAFKA_CLUSTERS_0_ENABLESR=false \\\n",
    "  -e KAFKA_CLUSTERS_0_SASLMECHANISM= \\\n",
    "  -e KAFKA_CLUSTERS_0_SASLPLAIN_USERNAME= \\\n",
    "  -e KAFKA_CLUSTERS_0_SASLPLAIN_PASSWORD= \\\n",
    "  -e KAFKA_CLUSTERS_0_SASLPLAIN_PASSWORD_FILE= \\\n",
    "  -e KAFKA_CLUSTERS_0_TRUSTEDCERTS= \\\n",
    "  -e KAFKA_CLUSTERS_0_CLIENTCERT= \\\n",
    "  -e KAFKA_CLUSTERS_0_CLIENTKEY= \\\n",
    "  -e KAFKA_CLUSTERS_0_CLIENTKEYPASSWORD= \\\n",
    "  -e KAFKA_CLUSTERS_0_CONSUMERCONFIGS= \\\n",
    "  -e KAFKA_CLUSTERS_0_ADMINCONFIGS= \\\n",
    "  provectuslabs/kafka-ui:latest"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "78d8323c-96a3-424c-aa0e-6840626cbe25",
   "metadata": {},
   "source": [
    "<li>Create Elasticsearch docker image:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "cfd185d4-69d7-4d84-9af1-4b9166e5e3b1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "#1 [internal] load build definition from Dockerfile\n",
      "#1 sha256:9951727a17f9e69414b97f80d66b4b9ec4d41e089e5c8f0c7f43d5b91c0e43a2\n",
      "#1 transferring dockerfile: 218B done\n",
      "#1 DONE 0.0s\n",
      "\n",
      "#2 [internal] load .dockerignore\n",
      "#2 sha256:1291cf5f600c667f3ade4c49609fdf8ee5e289e579fbae28077be7abb72a4d5e\n",
      "#2 transferring context: 2B done\n",
      "#2 DONE 0.0s\n",
      "\n",
      "#3 [internal] load metadata for docker.elastic.co/elasticsearch/elasticsearch:7.17.0\n",
      "#3 sha256:1dbbb84f972fa6e5ae3847ac9fa346a7bfdb27185aacf0e30b34e90ee821bb02\n",
      "#3 DONE 1.6s\n",
      "\n",
      "#5 [1/2] FROM docker.elastic.co/elasticsearch/elasticsearch:7.17.0@sha256:577b382dda5d05385aea8c7b60dad97e02ff41ca0da54f723151c2aed9ac8f54\n",
      "#5 sha256:248baad951fdafcfd3b149db719c51cb71d55a7d2b66d874122e699a682cfd13\n",
      "#5 DONE 0.0s\n",
      "\n",
      "#4 [2/2] RUN sysctl -w vm.max_map_count=262144\n",
      "#4 sha256:d2e7b07617b1fb5c3a397425da5323d5228d4474b3458426eed823b9c1ad013b\n",
      "#4 CACHED\n",
      "\n",
      "#6 exporting to image\n",
      "#6 sha256:e8c613e07b0b7ff33893b694f7759a10d42e180f2b4dc349fb57dc6b71dcab00\n",
      "#6 exporting layers done\n",
      "#6 writing image sha256:1bb3cf1b2633a08bffa3731f746200ad2fc13be52b90a11ea61d287d2820532a done\n",
      "#6 naming to docker.io/library/elastic-image done\n",
      "#6 DONE 0.0s\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker build -t elastic-image ./elasticsearch"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05b4575b-4fd9-4b9c-9e5c-0a079c94e204",
   "metadata": {},
   "source": [
    "<li>Then run Elasticsearch:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7bfcc177",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1d4dde6207608fcad1a02855d084de5b5d73fa028639929e92594d0562486637\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d --name elasticsearch --network kafka-network -p 9200:9200 -p 9300:9300 -e \"discovery.type=single-node\" elastic-image"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5f2a139c-37f3-4398-b498-67c0529af2a8",
   "metadata": {},
   "source": [
    "<li>Create a Kibana container:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "28bffbdc-61e6-4b97-99de-980fc83987c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "88e1763732cc40439a46fa1c20cd96f3b8e8708112222b075e0466c702596ea0\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d --name kibana -p 5601:5601 --network kafka-network -e \"ELASTICSEARCH_HOSTS=http://elasticsearch:9200\" docker.elastic.co/kibana/kibana:7.15.1"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ee9857ba-39f4-4c98-9a33-1fbd35a4ea33",
   "metadata": {},
   "source": [
    "<li>Run python script:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "5e8f7c86-bdb9-439e-a3e6-a3ca27f9065c",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "python3 ./logstash/urls_dates.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b8c4d72-713b-4bc4-ba6c-83bf1b572eae",
   "metadata": {},
   "source": [
    "<li>Create Logstash docker image:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "958290c0-50ed-4c19-b88e-e146c5fc0582",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "#1 [internal] load build definition from Dockerfile\n",
      "#1 sha256:2bc9b4850f0190c69af6131e677837799905b5bdd2c9c71e4f502a5b40fea358\n",
      "#1 transferring dockerfile: 520B 0.0s done\n",
      "#1 DONE 0.1s\n",
      "\n",
      "#2 [internal] load .dockerignore\n",
      "#2 sha256:200dc0774b78b25e34bd7b537a7d4121eb068a37cd04c603ce49cfd70bb935a0\n",
      "#2 transferring context:\n",
      "#2 transferring context: 2B done\n",
      "#2 DONE 0.1s\n",
      "\n",
      "#3 [internal] load metadata for docker.elastic.co/logstash/logstash:8.8.1@sha256:9b2e080605e208ef1165fd6cfd68a8b05c2031c8818b8520f82f73238dbb471c\n",
      "#3 sha256:f9b51405f81dc3157b837175d3f8cd610533f41e65fe5a83aa380f95b96c6a5c\n",
      "#3 DONE 1.4s\n",
      "\n",
      "#4 [1/6] FROM docker.elastic.co/logstash/logstash:8.8.1@sha256:9b2e080605e208ef1165fd6cfd68a8b05c2031c8818b8520f82f73238dbb471c\n",
      "#4 sha256:1001cfd24a82b78a0aded882f8f22066a01463643e26780d6e6428154760017a\n",
      "#4 DONE 0.0s\n",
      "\n",
      "#5 [internal] load build context\n",
      "#5 sha256:82a405591fcdbb6a6f5abc96202926f06750d4cd62cec3224368fe7e26824e64\n",
      "#5 transferring context: 1.05MB done\n",
      "#5 DONE 0.0s\n",
      "\n",
      "#6 [2/6] COPY logstash.conf /usr/share/logstash/pipeline/logstash.conf\n",
      "#6 sha256:78e132c819f36c00a9424f2499f6e282324047425537ac028ab7d4fed1f97379\n",
      "#6 CACHED\n",
      "\n",
      "#7 [3/6] COPY seismic_portal_links.txt /usr/share/logstash/seismic_portal_links.txt\n",
      "#7 sha256:b63ef3471ce3ded624792217f6fd5d39cd77665b9e81a1fadbd5fd934441dd96\n",
      "#7 CACHED\n",
      "\n",
      "#8 [4/6] RUN logstash-plugin install logstash-filter-xml\n",
      "#8 sha256:7e07136d6dff7f9d62aa1301264ff13351c539b081ccd26313f75586d9eb8f8e\n",
      "#8 CACHED\n",
      "\n",
      "#9 [5/6] RUN logstash-plugin install logstash-filter-mutate\n",
      "#9 sha256:661e07d757a805ebae1b828bdd136990695e1f0f5ae2c1f873c504cb51a02d81\n",
      "#9 CACHED\n",
      "\n",
      "#10 [6/6] RUN logstash-plugin install logstash-input-http\n",
      "#10 sha256:f4baa3177c0e5991e8c1c20ced22bb74d3ee8544fde50f256b5b6c1e4054c334\n",
      "#10 CACHED\n",
      "\n",
      "#11 exporting to image\n",
      "#11 sha256:e8c613e07b0b7ff33893b694f7759a10d42e180f2b4dc349fb57dc6b71dcab00\n",
      "#11 exporting layers done\n",
      "#11 writing image sha256:9f91bde7510a875c6b3c876c114ab153bb42acf821f265bf9a7e5cfe914fde9a\n",
      "#11 writing image sha256:9f91bde7510a875c6b3c876c114ab153bb42acf821f265bf9a7e5cfe914fde9a done\n",
      "#11 naming to docker.io/library/logstash-image done\n",
      "#11 DONE 0.1s\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker build -t logstash-image ./logstash"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0e59e0ef-39d9-4f28-a90b-4ea78df6be9b",
   "metadata": {},
   "source": [
    "<li>Then run Logstash:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "35060c65-e8ff-4adf-bc30-7759bb322e03",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "18538312611fae3ca7509e117afc3db296f2da92e326a4f5ff40a749bf3447ad\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d --name logstash-container --network kafka-network  logstash-image"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3baa7ec7-b449-4a7f-937f-ec7eb8e956dd",
   "metadata": {},
   "source": [
    "<li>Run following code for download elasticsearch-spark</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "41b69312-1947-490d-9c31-9963bcab4ef5",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "--2023-07-28 00:38:47--  https://repo1.maven.org/maven2/org/elasticsearch/elasticsearch-spark-20_2.12/7.15.1/elasticsearch-spark-20_2.12-7.15.1.jar\n",
      "Resolving repo1.maven.org (repo1.maven.org)... 199.232.192.209, 199.232.196.209, 2a04:4e42:4c::209, ...\n",
      "Connecting to repo1.maven.org (repo1.maven.org)|199.232.192.209|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 2101583 (2.0M) [application/java-archive]\n",
      "Saving to: ‘./spark/elasticsearch-spark-20_2.12-7.15.1.jar’\n",
      "\n",
      "     0K .......... .......... .......... .......... ..........  2% 2.96M 1s\n",
      "    50K .......... .......... .......... .......... ..........  4% 5.90M 0s\n",
      "   100K .......... .......... .......... .......... ..........  7% 4.97M 0s\n",
      "   150K .......... .......... .......... .......... ..........  9% 17.2M 0s\n",
      "   200K .......... .......... .......... .......... .......... 12% 73.1M 0s\n",
      "   250K .......... .......... .......... .......... .......... 14% 5.20M 0s\n",
      "   300K .......... .......... .......... .......... .......... 17% 43.6M 0s\n",
      "   350K .......... .......... .......... .......... .......... 19% 10.0M 0s\n",
      "   400K .......... .......... .......... .......... .......... 21% 27.0M 0s\n",
      "   450K .......... .......... .......... .......... .......... 24%  190M 0s\n",
      "   500K .......... .......... .......... .......... .......... 26%  293M 0s\n",
      "   550K .......... .......... .......... .......... .......... 29%  324M 0s\n",
      "   600K .......... .......... .......... .......... .......... 31% 5.52M 0s\n",
      "   650K .......... .......... .......... .......... .......... 34%  177M 0s\n",
      "   700K .......... .......... .......... .......... .......... 36%  324M 0s\n",
      "   750K .......... .......... .......... .......... .......... 38%  153M 0s\n",
      "   800K .......... .......... .......... .......... .......... 41%  149M 0s\n",
      "   850K .......... .......... .......... .......... .......... 43% 37.7M 0s\n",
      "   900K .......... .......... .......... .......... .......... 46%  101M 0s\n",
      "   950K .......... .......... .......... .......... .......... 48%  105M 0s\n",
      "  1000K .......... .......... .......... .......... .......... 51% 67.8M 0s\n",
      "  1050K .......... .......... .......... .......... .......... 53%  168M 0s\n",
      "  1100K .......... .......... .......... .......... .......... 56% 63.0M 0s\n",
      "  1150K .......... .......... .......... .......... .......... 58%  300M 0s\n",
      "  1200K .......... .......... .......... .......... .......... 60% 11.4M 0s\n",
      "  1250K .......... .......... .......... .......... .......... 63% 16.7M 0s\n",
      "  1300K .......... .......... .......... .......... .......... 65% 80.7M 0s\n",
      "  1350K .......... .......... .......... .......... .......... 68%  183M 0s\n",
      "  1400K .......... .......... .......... .......... .......... 70% 50.8M 0s\n",
      "  1450K .......... .......... .......... .......... .......... 73%  157M 0s\n",
      "  1500K .......... .......... .......... .......... .......... 75%  310M 0s\n",
      "  1550K .......... .......... .......... .......... .......... 77% 19.8M 0s\n",
      "  1600K .......... .......... .......... .......... .......... 80%  164M 0s\n",
      "  1650K .......... .......... .......... .......... .......... 82%  314M 0s\n",
      "  1700K .......... .......... .......... .......... .......... 85%  192M 0s\n",
      "  1750K .......... .......... .......... .......... .......... 87%  217M 0s\n",
      "  1800K .......... .......... .......... .......... .......... 90%  163M 0s\n",
      "  1850K .......... .......... .......... .......... .......... 92%  189M 0s\n",
      "  1900K .......... .......... .......... .......... .......... 95%  313M 0s\n",
      "  1950K .......... .......... .......... .......... .......... 97%  320M 0s\n",
      "  2000K .......... .......... .......... .......... .......... 99%  269M 0s\n",
      "  2050K ..                                                    100% 4439G=0.08s\n",
      "\n",
      "2023-07-28 00:38:47 (23.9 MB/s) - ‘./spark/elasticsearch-spark-20_2.12-7.15.1.jar’ saved [2101583/2101583]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "wget https://repo1.maven.org/maven2/org/elasticsearch/elasticsearch-spark-20_2.12/7.15.1/elasticsearch-spark-20_2.12-7.15.1.jar -P ./spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dca2b6f4-29e2-4710-8479-38f0451301bf",
   "metadata": {},
   "source": [
    "<li>And spark-sql-kafka:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "1d5bf531-9c5b-4915-a8c4-38ca6c412cbc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "--2023-07-28 00:38:47--  https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.4.1/spark-sql-kafka-0-10_2.12-3.4.1.jar\n",
      "Resolving repo1.maven.org (repo1.maven.org)... 199.232.192.209, 199.232.196.209, 2a04:4e42:4c::209, ...\n",
      "Connecting to repo1.maven.org (repo1.maven.org)|199.232.192.209|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 427253 (417K) [application/java-archive]\n",
      "Saving to: ‘./spark/spark-sql-kafka-0-10_2.12-3.4.1.jar’\n",
      "\n",
      "     0K .......... .......... .......... .......... .......... 11% 2.92M 0s\n",
      "    50K .......... .......... .......... .......... .......... 23% 3.93M 0s\n",
      "   100K .......... .......... .......... .......... .......... 35% 4.50M 0s\n",
      "   150K .......... .......... .......... .......... .......... 47% 30.7M 0s\n",
      "   200K .......... .......... .......... .......... .......... 59% 8.93M 0s\n",
      "   250K .......... .......... .......... .......... .......... 71%  307M 0s\n",
      "   300K .......... .......... .......... .......... .......... 83% 98.3M 0s\n",
      "   350K .......... .......... .......... .......... .......... 95% 5.22M 0s\n",
      "   400K .......... .......                                    100%  199M=0.06s\n",
      "\n",
      "2023-07-28 00:38:47 (7.13 MB/s) - ‘./spark/spark-sql-kafka-0-10_2.12-3.4.1.jar’ saved [427253/427253]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "wget https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.4.1/spark-sql-kafka-0-10_2.12-3.4.1.jar -P ./spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "833f7de0-70a6-4be7-9554-53452beb1b1b",
   "metadata": {},
   "source": [
    "<li>Create Apache Spark docker image:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "78daf47c-d945-4b16-9d4d-5714fc8c0872",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "#1 [internal] load build definition from Dockerfile\n",
      "#1 sha256:841f7323cfafd9bcfe25788479f6fada23565f9dd581ada3649a608ea09da417\n",
      "#1 transferring dockerfile: 297B done\n",
      "#1 DONE 0.1s\n",
      "\n",
      "#2 [internal] load .dockerignore\n",
      "#2 sha256:1b9edfe5fc3d400dd16b156260f9d9be7846c010f1f4632c849ca56525d6e4e7\n",
      "#2 transferring context: 2B done\n",
      "#2 DONE 0.1s\n",
      "\n",
      "#3 [internal] load metadata for docker.io/bitnami/spark:latest\n",
      "#3 sha256:0dbae0baa930a0ea10c2e1420f888b307c97531e436e834af50ecdbfd42c80cf\n",
      "#3 DONE 1.3s\n",
      "\n",
      "#4 [1/6] FROM docker.io/bitnami/spark:latest@sha256:9467c6ec2cfd0cde0cb23ea81f44f85430cb0a8154d8a06982ec8895b1734b00\n",
      "#4 sha256:77a30c37a56681f4cc8c11e90dd833687639c5598682c3f6884ea1d4a066bc22\n",
      "#4 DONE 0.0s\n",
      "\n",
      "#6 [internal] load build context\n",
      "#6 sha256:9c850b7898abbb14c3b4abe5079f4de386f93eff755d0618db99490029b4090d\n",
      "#6 transferring context: 5.28kB done\n",
      "#6 DONE 0.0s\n",
      "\n",
      "#8 [4/6] COPY requirements.txt .\n",
      "#8 sha256:fd34911b55807c0364036a39cb923f67d24c7859b8538520b717bfd081aa57b8\n",
      "#8 CACHED\n",
      "\n",
      "#7 [3/6] COPY earthquake_analysis.py .\n",
      "#7 sha256:70f9fa76892c412ed01cf556118cba946b0dd41322ac88ecf9b288ec847c89ef\n",
      "#7 CACHED\n",
      "\n",
      "#5 [2/6] WORKDIR /app\n",
      "#5 sha256:20e4ff9581b8eb2c66f8d3294eb22a5030843d3a149e06728d535f1c52ef4d3b\n",
      "#5 CACHED\n",
      "\n",
      "#9 [5/6] COPY elasticsearch-spark-20_2.12-7.15.1.jar /opt/bitnami/spark/jars/\n",
      "#9 sha256:e0e18ee88ab7fd3eabd90df36906f4600ced84296eb582c94e4b4b874b3bfacd\n",
      "#9 CACHED\n",
      "\n",
      "#10 [6/6] RUN pip install -r requirements.txt\n",
      "#10 sha256:3330effd7752f78ee6951a7262c640bf4918d3c5b970ed12d56f1f7e3b2f0850\n",
      "#10 CACHED\n",
      "\n",
      "#11 exporting to image\n",
      "#11 sha256:e8c613e07b0b7ff33893b694f7759a10d42e180f2b4dc349fb57dc6b71dcab00\n",
      "#11 exporting layers done\n",
      "#11 writing image sha256:41bdaa3cea58ab5cad2425e10b43ca4c0a4cad42b3d71ca3434966d66fc557f2\n",
      "#11 writing image sha256:41bdaa3cea58ab5cad2425e10b43ca4c0a4cad42b3d71ca3434966d66fc557f2 0.0s done\n",
      "#11 naming to docker.io/library/spark-earthquakes done\n",
      "#11 DONE 0.0s\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker build -t spark-earthquakes ./spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "decada74-cf98-4356-8ba0-cbb3514c0a71",
   "metadata": {},
   "source": [
    "<li>Wait until all messages in Kafka's \"earthquakes\" topic are ready, then run Apache Spark container:</li>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "d2ed0ce5-2088-4ca1-9efa-b50176450d82",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "c70546e50bb806f6ffb49f04295b0ee10788a82c38927551b897c5a61ae98441\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    ". ~/.bashrc\n",
    "docker run -d --network kafka-network --name spark-earthquakes_analyzer spark-earthquakes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "78e925e9-95f5-4c04-9309-d9458b93b5b6",
   "metadata": {},
   "source": [
    "<li>In the browser, put the url \"http://localhost:5601\", go to \"Kibana / Saved Objects\", click on \"import\" and select \"export.ndjson\" in kibana folder</li>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bbb02840-18dd-4137-bbaf-8a4984ba1a82",
   "metadata": {},
   "source": [
    "<li>Open left menu in Kibana, select Dashboard, select the imported dashboard for see all lens</li>"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}