pyFlink_Astra_batch.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "c1827610-1dbd-49f1-ad86-a8f54e4f0edd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyflink.datastream import StreamExecutionEnvironment\n",
    "import requests\n",
    "from pyflink.table import StreamTableEnvironment, TableDescriptor, Schema, DataTypes\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "9562e1f7-8374-4716-8114-2b18811e3166",
   "metadata": {},
   "outputs": [],
   "source": [
    "from my_local_secrets import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "88b263ce-fa74-4818-87df-197b8f2e392b",
   "metadata": {},
   "outputs": [],
   "source": [
    "symbol = \"IBM\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae004f56-fe36-4e1d-b195-664083b0a0d3",
   "metadata": {},
   "source": [
    "## Defining the API query params and function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ec638841-0b87-404a-b801-2f3654bd7bb4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def fetch_data(api_key, symbol):\n",
    "    url = f\"https://www.alphavantage.co/query?function=TIME_SERIES_INTRADAY&symbol={symbol}&interval=5min&apikey={api_key}\"\n",
    "    response = requests.get(url)\n",
    "    data = json.loads(response.text)\n",
    "    return data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bd700664-14fe-42ca-a68a-d8c70721494e",
   "metadata": {},
   "source": [
    "## Preparing the pyFlink environment and setting the source for our collection of data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "db0162c1-33bd-4e21-a26c-a40e0e247f6a",
   "metadata": {},
   "outputs": [],
   "source": [
    "env = StreamExecutionEnvironment.get_execution_environment()\n",
    "t_env = StreamTableEnvironment.create(stream_execution_environment=env)\n",
    "\n",
    "ds = env.from_collection([fetch_data(api_key, symbol)])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "39282cf2-8262-408a-91ce-9b87d6725ad1",
   "metadata": {},
   "source": [
    "## Processing the json object and filter it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "50f16e28-1786-4c40-aea0-b0fbe0a26339",
   "metadata": {},
   "outputs": [],
   "source": [
    "import itertools\n",
    "## Keep only records with volume higher than 100k\n",
    "def update_tel(data):\n",
    "    all_values=[]\n",
    "    for elt in data['Time Series (5min)']:\n",
    "        all_values.append(list(itertools.chain(*[[elt], list(data['Time Series (5min)'][elt].values())])))\n",
    "    return [item for item in all_values if int(item[5]) > 100000]\n",
    "    \n",
    "ds = ds.map(update_tel)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "5b83359b-1594-4681-b46a-3da140741649",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyflink.datastream.data_stream.DataStreamSink at 0x1180742e0>"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ds.print()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "326905ef-b7e6-4916-a281-8bd78146e29d",
   "metadata": {},
   "source": [
    "### Prepare the DB part: create the table to use for data insertion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5788d0c2-ec02-41c8-afcc-9b85842ce6ed",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x16a1db400>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from cassandra.cluster import Cluster\n",
    "from cassandra.auth import PlainTextAuthProvider\n",
    "## Astra Configuration\n",
    "cloud_config= {\n",
    "  'secure_connect_bundle': secure_bundle_path\n",
    "}\n",
    "auth_provider = PlainTextAuthProvider(client_id, client_secret)\n",
    "cluster = Cluster(cloud=cloud_config, auth_provider=auth_provider)\n",
    "session = cluster.connect()\n",
    "\n",
    "session.execute(f'USE {db_keyspace}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "e0c7bb5c-e485-4756-a855-b7c1f5a7ea1d",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<cassandra.cluster.ResultSet at 0x16a0073a0>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "query = \"\"\"\n",
    "CREATE TABLE if not exists market_stock_data (\n",
    "    date text,\n",
    "    open float,\n",
    "    high float,\n",
    "    low float,\n",
    "    close float,\n",
    "    volume float,\n",
    "    PRIMARY KEY (date)\n",
    ")\n",
    "\"\"\"\n",
    "session.execute(query)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11d01594-0d22-433d-9ace-db15a6e8cf5a",
   "metadata": {},
   "source": [
    "## Use the AstraDB RestAPI for data insertion via pyFlink"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "141e6721-38dc-45fe-b6d9-0af5e7b63628",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyflink.datastream.data_stream.DataStreamSink at 0x10a74d870>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def send_to_rest_api(data):\n",
    "    url = f'https://{astra_id}-{astra_region}.apps.astra.datastax.com/api/rest/v2/keyspaces/{db_keyspace}/market_stock_data'\n",
    "    print(url)\n",
    "    headers = {'content-type': 'application/json', 'x-cassandra-token': token}\n",
    "    for row in data: \n",
    "        response = requests.post(url, headers=headers, json=json.loads('{\"date\": \"' + str(row[0]) + '\", \"open\": \"' + row[1] + '\", \"high\": \"' + row[2] + '\", \"low\": \"' + row[3] +  '\", \"close\": \"' + row[4] + '\", \"volume\": \"' + row[5] + '\"}'), verify=False)  \n",
    "        if response.status_code == 200 or response.status_code == 201:\n",
    "            print(\"Data sent successfully to REST API\")\n",
    "        else:\n",
    "            print(f\"Failed to send data to REST API. Error: {response.text}\")\n",
    "\n",
    "# Create a DataStream from the collection\n",
    "ds.map(lambda x: send_to_rest_api(x))\n",
    "\n",
    "ds.print()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1b55cbc-2b56-480c-a4cb-dae47ea4d9d1",
   "metadata": {},
   "source": [
    "## Execute our pyFlink graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "0abbc8ac-b9fb-4529-912e-f19520339d88",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "https://d5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com/api/rest/v2/keyspaces/tf_keyspace/market_stock_data\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/homebrew/Caskroom/miniconda/base/lib/python3.10/site-packages/urllib3/connectionpool.py:1056: InsecureRequestWarning: Unverified HTTPS request is being made to host 'd5103402-d2c2-49ee-8406-0f7edcf01fed-us-east1.apps.astra.datastax.com'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data sent successfully to REST API\n",
      "4> [['2023-07-10 19:00:00', '132.9000', '132.9000', '132.7500', '132.7500', '293396'], ['2023-07-10 18:30:00', '132.9000', '132.9000', '132.9000', '132.9000', '293411'], ['2023-07-10 16:10:00', '132.9000', '132.9440', '132.9000', '132.9440', '293395'], ['2023-07-10 16:00:00', '132.9300', '133.0200', '132.6300', '133.0000', '632125'], ['2023-07-10 15:55:00', '132.8000', '132.9400', '132.7300', '132.9400', '227333'], ['2023-07-10 15:50:00', '132.6900', '132.8200', '132.6050', '132.8000', '113843']]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<pyflink.common.job_execution_result.JobExecutionResult at 0x1080e5db0>"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "env.execute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc095c84-c41b-4104-ba66-a098df415f45",
   "metadata": {},
   "source": [
    "### Check the data in the AstraDB table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "ba2ac203-7a92-4fb2-be20-3dde402bfcc5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>date</th>\n",
       "      <th>close</th>\n",
       "      <th>high</th>\n",
       "      <th>low</th>\n",
       "      <th>open</th>\n",
       "      <th>volume</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2023-07-10 16:00:00</td>\n",
       "      <td>133.000000</td>\n",
       "      <td>133.020004</td>\n",
       "      <td>132.630005</td>\n",
       "      <td>132.929993</td>\n",
       "      <td>632125.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2023-07-10 18:30:00</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>293411.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2023-07-10 16:10:00</td>\n",
       "      <td>132.944000</td>\n",
       "      <td>132.944000</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>293395.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2023-07-06 19:00:00</td>\n",
       "      <td>132.100006</td>\n",
       "      <td>132.160004</td>\n",
       "      <td>132.100006</td>\n",
       "      <td>132.160004</td>\n",
       "      <td>606639.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2023-07-10 15:50:00</td>\n",
       "      <td>132.800003</td>\n",
       "      <td>132.820007</td>\n",
       "      <td>132.604996</td>\n",
       "      <td>132.690002</td>\n",
       "      <td>113843.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2023-07-10 15:55:00</td>\n",
       "      <td>132.940002</td>\n",
       "      <td>132.940002</td>\n",
       "      <td>132.729996</td>\n",
       "      <td>132.800003</td>\n",
       "      <td>227333.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>2023-07-10 19:00:00</td>\n",
       "      <td>132.750000</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>132.750000</td>\n",
       "      <td>132.899994</td>\n",
       "      <td>293396.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                  date       close        high         low        open  \\\n",
       "0  2023-07-10 16:00:00  133.000000  133.020004  132.630005  132.929993   \n",
       "1  2023-07-10 18:30:00  132.899994  132.899994  132.899994  132.899994   \n",
       "2  2023-07-10 16:10:00  132.944000  132.944000  132.899994  132.899994   \n",
       "3  2023-07-06 19:00:00  132.100006  132.160004  132.100006  132.160004   \n",
       "4  2023-07-10 15:50:00  132.800003  132.820007  132.604996  132.690002   \n",
       "5  2023-07-10 15:55:00  132.940002  132.940002  132.729996  132.800003   \n",
       "6  2023-07-10 19:00:00  132.750000  132.899994  132.750000  132.899994   \n",
       "\n",
       "     volume  \n",
       "0  632125.0  \n",
       "1  293411.0  \n",
       "2  293395.0  \n",
       "3  606639.0  \n",
       "4  113843.0  \n",
       "5  227333.0  \n",
       "6  293396.0  "
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "pd.DataFrame(session.execute(\"select * from tf_keyspace.market_stock_data;\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7959f2ab-0cac-4550-9b8d-e91358683906",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "53584720-66f6-405d-86ca-d66ded0558b4",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}