02 Connection Streams Tables.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "890fb533",
   "metadata": {},
   "source": [
    "# Step 1: Create twitter connection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d94eeed6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100  1587  100   781  100   806   9409   9710 --:--:-- --:--:-- --:--:-- 19120\n",
      "{\n",
      "   \"config\" : {\n",
      "      \"connector.class\" : \"com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector\",\n",
      "      \"filter.keywords\" : \"covid, covid-19, vaccine\",\n",
      "      \"kafka.status.topic\" : \"covidtweets\",\n",
      "      \"name\" : \"TwitterConnector\",\n",
      "      \"process.deletes\" : \"true\",\n",
      "      \"quickstart\" : \"covidtweets\",\n",
      "      \"tasks.max\" : \"1\",\n",
      "      \"topic.creation.default.partitions\" : \"1\",\n",
      "      \"topic.creation.default.replication.factor\" : \"1\",\n",
      "      \"topic.creation.groups\" : \"covidtweets\",\n",
      "      \"twitter.debug\" : \"true\",\n",
      "      \"twitter.oauth.accessToken\" : \"YOUR_CONSUMER_ACCESS_TOKEN\",\n",
      "      \"twitter.oauth.accessTokenSecret\" : \"YOUR_CONSUMER_ACCESS_TOKEN_SECRET\",\n",
      "      \"twitter.oauth.consumerKey\" : \"YOUR_CONSUMER_KEY\",\n",
      "      \"twitter.oauth.consumerSecret\" : \"YOUR_CONSUMER_SECRET\"\n",
      "   },\n",
      "   \"name\" : \"TwitterConnector\",\n",
      "   \"tasks\" : [],\n",
      "   \"type\" : \"source\"\n",
      "}\n"
     ]
    }
   ],
   "source": [
    "#It creates the Twitter API connection and the topic (covidtweets) to pass the information.\n",
    "!curl -X POST -H \"Content-Type: application/json\" --data @configs/twitterConnector.json http://connect:8083/connectors | json_pp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3ab508c2",
   "metadata": {},
   "source": [
    "# Step 2: Create streams (ksql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "7647ef62",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100  3148  100  3026  100   122   6940    279 --:--:-- --:--:-- --:--:--  7203\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"stream/`RAWTWEETS`/create\",\n",
      "      \"commandSequenceNumber\" : 2,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Stream created\",\n",
      "         \"queryId\" : null,\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE STREAM RAWTWEETS (CREATEDAT BIGINT, ID BIGINT, TEXT STRING, SOURCE STRING, TRUNCATED BOOLEAN, INREPLYTOSTATUSID BIGINT, INREPLYTOUSERID BIGINT, INREPLYTOSCREENNAME STRING, GEOLOCATION STRUCT<LATITUDE DOUBLE, LONGITUDE DOUBLE>, PLACE STRUCT<NAME STRING, STREETADDRESS STRING, COUNTRYCODE STRING, ID STRING, COUNTRY STRING, PLACETYPE STRING, URL STRING, FULLNAME STRING>, FAVORITED BOOLEAN, RETWEETED BOOLEAN, FAVORITECOUNT INTEGER, USER STRUCT<ID BIGINT, NAME STRING, SCREENNAME STRING, LOCATION STRING, DESCRIPTION STRING, CONTRIBUTORSENABLED BOOLEAN, PROFILEIMAGEURL STRING, BIGGERPROFILEIMAGEURL STRING, MINIPROFILEIMAGEURL STRING, ORIGINALPROFILEIMAGEURL STRING, PROFILEIMAGEURLHTTPS STRING, BIGGERPROFILEIMAGEURLHTTPS STRING, MINIPROFILEIMAGEURLHTTPS STRING, ORIGINALPROFILEIMAGEURLHTTPS STRING, DEFAULTPROFILEIMAGE BOOLEAN, URL STRING, PROTECTED BOOLEAN, FOLLOWERSCOUNT INTEGER, PROFILEBACKGROUNDCOLOR STRING, PROFILETEXTCOLOR STRING, PROFILELINKCOLOR STRING, PROFILESIDEBARFILLCOLOR STRING, PROFILESIDEBARBORDERCOLOR STRING, PROFILEUSEBACKGROUNDIMAGE BOOLEAN, DEFAULTPROFILE BOOLEAN, SHOWALLINLINEMEDIA BOOLEAN, FRIENDSCOUNT INTEGER, CREATEDAT BIGINT, FAVOURITESCOUNT INTEGER, UTCOFFSET INTEGER, TIMEZONE STRING, PROFILEBACKGROUNDIMAGEURL STRING, PROFILEBACKGROUNDIMAGEURLHTTPS STRING, PROFILEBANNERURL STRING, PROFILEBANNERRETINAURL STRING, PROFILEBANNERIPADURL STRING, PROFILEBANNERIPADRETINAURL STRING, PROFILEBANNERMOBILEURL STRING, PROFILEBANNERMOBILERETINAURL STRING, PROFILEBACKGROUNDTILED BOOLEAN, LANG STRING, STATUSESCOUNT INTEGER, GEOENABLED BOOLEAN, VERIFIED BOOLEAN, TRANSLATOR BOOLEAN, LISTEDCOUNT INTEGER, FOLLOWREQUESTSENT BOOLEAN, WITHHELDINCOUNTRIES ARRAY<STRING>>, RETWEET BOOLEAN, CONTRIBUTORS ARRAY<BIGINT>, RETWEETCOUNT INTEGER, RETWEETEDBYME BOOLEAN, CURRENTUSERRETWEETID BIGINT, POSSIBLYSENSITIVE BOOLEAN, LANG STRING, WITHHELDINCOUNTRIES ARRAY<STRING>, HASHTAGENTITIES ARRAY<STRUCT<TEXT STRING, START INTEGER, `END` INTEGER>>, USERMENTIONENTITIES ARRAY<STRUCT<NAME STRING, ID BIGINT, TEXT STRING, SCREENNAME STRING, START INTEGER, `END` INTEGER>>, MEDIAENTITIES ARRAY<STRUCT<ID BIGINT, TYPE STRING, MEDIAURL STRING, SIZES MAP<STRING, STRUCT<RESIZE INTEGER, WIDTH INTEGER, HEIGHT INTEGER>>, MEDIAURLHTTPS STRING, VIDEOASPECTRATIOWIDTH INTEGER, VIDEOASPECTRATIOHEIGHT INTEGER, VIDEODURATIONMILLIS BIGINT, VIDEOVARIANTS ARRAY<STRUCT<URL STRING, BITRATE INTEGER, CONTENTTYPE STRING>>, EXTALTTEXT STRING, URL STRING, TEXT STRING, EXPANDEDURL STRING, START INTEGER, `END` INTEGER, DISPLAYURL STRING>>, SYMBOLENTITIES ARRAY<STRUCT<START INTEGER, `END` INTEGER, TEXT STRING>>, URLENTITIES ARRAY<STRUCT<URL STRING, TEXT STRING, EXPANDEDURL STRING, START INTEGER, `END` INTEGER, DISPLAYURL STRING>>) WITH (KAFKA_TOPIC='covidtweets', KEY_FORMAT='KAFKA', VALUE_FORMAT='AVRO', VALUE_SCHEMA_ID=1);\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates the data stream form the covidtweets topic. It is called RAWTWEETS\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/RawStream.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "3ce93f77",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100  2007  100  1239  100   768   1002    621  0:00:01  0:00:01 --:--:--  1623\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"stream/`ENGSTREAM`/create\",\n",
      "      \"commandSequenceNumber\" : 4,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Created query with ID CSAS_ENGSTREAM_3\",\n",
      "         \"queryId\" : \"CSAS_ENGSTREAM_3\",\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE STREAM ENGSTREAM WITH (KAFKA_TOPIC='engtweets', PARTITIONS=1, REPLICAS=1, VALUE_FORMAT='JSON') AS SELECT\\n  RAWTWEETS.CREATEDAT CREATEDAT,\\n  RAWTWEETS.ID ID,\\n  RAWTWEETS.USER->ID USER_ID,\\n  RAWTWEETS.USER->NAME USER_NAME,\\n  RAWTWEETS.USER->SCREENNAME SCREENNAME,\\n  RAWTWEETS.USER->LOCATION USER_LOCATION,\\n  RAWTWEETS.USER->FOLLOWERSCOUNT FOLLOW_COUNT,\\n  RAWTWEETS.USER->FRIENDSCOUNT FRIEND_COUNT,\\n  RAWTWEETS.USER->CREATEDAT USER_CREAT_AT,\\n  RAWTWEETS.USER->FAVOURITESCOUNT USER_FAV_COUNT,\\n  RAWTWEETS.USER->VERIFIED USER_VERIFIED,\\n  RAWTWEETS.USER->LANG USER_LANG,\\n  RAWTWEETS.TEXT TEXT,\\n  RAWTWEETS.LANG LANG,\\n  RAWTWEETS.GEOLOCATION->LATITUDE GEO_LAT,\\n  RAWTWEETS.GEOLOCATION->LONGITUDE GEO_LONG,\\n  RAWTWEETS.PLACE->NAME PLACE_NAME,\\n  RAWTWEETS.PLACE->COUNTRY PLACE_COUNTRY,\\n  RAWTWEETS.HASHTAGENTITIES HASHTAG,\\n  RAWTWEETS.USERMENTIONENTITIES USER_MENTION_NAME\\nFROM RAWTWEETS RAWTWEETS\\nWHERE ((RAWTWEETS.LANG = 'en') AND (RAWTWEETS.RETWEET = false))\\nEMIT CHANGES;\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates a  stream from the RAWTWEETS. It selects some of the features and filters lang = eng. It is called ENGSTREAM\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/EngStream.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "863a3666",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100  1989  100  1223  100   766   2412   1510 --:--:-- --:--:-- --:--:--  3923\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"stream/`LOCSTREAM`/create\",\n",
      "      \"commandSequenceNumber\" : 18,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Created query with ID CSAS_LOCSTREAM_17\",\n",
      "         \"queryId\" : \"CSAS_LOCSTREAM_17\",\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE STREAM LOCSTREAM WITH (KAFKA_TOPIC='loctweets', PARTITIONS=1, REPLICAS=1, VALUE_FORMAT='JSON') AS SELECT\\n  RAWTWEETS.CREATEDAT CREATEDAT,\\n  RAWTWEETS.ID ID,\\n  RAWTWEETS.USER->ID USER_ID,\\n  RAWTWEETS.USER->NAME USER_NAME,\\n  RAWTWEETS.USER->SCREENNAME SCREENNAME,\\n  RAWTWEETS.USER->LOCATION USER_LOCATION,\\n  RAWTWEETS.USER->FOLLOWERSCOUNT FOLLOW_COUNT,\\n  RAWTWEETS.USER->FRIENDSCOUNT FRIEND_COUNT,\\n  RAWTWEETS.USER->CREATEDAT USER_CREAT_AT,\\n  RAWTWEETS.USER->FAVOURITESCOUNT USER_FAV_COUNT,\\n  RAWTWEETS.USER->VERIFIED USER_VERIFIED,\\n  RAWTWEETS.USER->LANG USER_LANG,\\n  RAWTWEETS.TEXT TEXT,\\n  RAWTWEETS.LANG LANG,\\n  RAWTWEETS.GEOLOCATION->LATITUDE GEO_LAT,\\n  RAWTWEETS.GEOLOCATION->LONGITUDE GEO_LONG,\\n  RAWTWEETS.PLACE->NAME PLACE_NAME,\\n  RAWTWEETS.PLACE->COUNTRY PLACE_COUNTRY,\\n  RAWTWEETS.HASHTAGENTITIES HASHTAG,\\n  RAWTWEETS.USERMENTIONENTITIES USER_MENTION_NAME\\nFROM RAWTWEETS RAWTWEETS\\nWHERE (RAWTWEETS.PLACE->COUNTRY IS NOT NULL)\\nEMIT CHANGES;\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates a stream from the RAWTWEETS. It selects some of the features and filters country is not null. It is called LOCSTREAM\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/LocStream.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c83182c6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100   649  100   458  100   191   1221    509 --:--:-- --:--:-- --:--:--  1726\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"stream/`HASHSTREAM`/create\",\n",
      "      \"commandSequenceNumber\" : 8,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Created query with ID CSAS_HASHSTREAM_7\",\n",
      "         \"queryId\" : \"CSAS_HASHSTREAM_7\",\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE STREAM HASHSTREAM WITH (KAFKA_TOPIC='HASHSTREAM', PARTITIONS=1, REPLICAS=1) AS SELECT\\n  RAWTWEETS.ID ID,\\n  LCASE(EXPLODE(RAWTWEETS.HASHTAGENTITIES)->TEXT) HASHTAG\\nFROM RAWTWEETS RAWTWEETS\\nEMIT CHANGES;\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates a stream from the RAWTWEETS. It lists the Hashtags on the tweets. It is called HASHSTREAM\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/HashStream.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e2fa397e",
   "metadata": {},
   "source": [
    "# Step 3: Create tables (ksql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "a8959cd3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100   655  100   488  100   167   1053    360 --:--:-- --:--:-- --:--:--  1414\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"table/`HASHCOUNT`/create\",\n",
      "      \"commandSequenceNumber\" : 10,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Created query with ID CTAS_HASHCOUNT_9\",\n",
      "         \"queryId\" : \"CTAS_HASHCOUNT_9\",\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE TABLE HASHCOUNT WITH (KAFKA_TOPIC='HASHCOUNT', PARTITIONS=1, REPLICAS=1) AS SELECT\\n  HASHSTREAM.HASHTAG HASHTAG,\\n  COUNT(*) CT\\nFROM HASHSTREAM HASHSTREAM\\nWINDOW TUMBLING ( SIZE 10 MINUTES ) \\nGROUP BY HASHSTREAM.HASHTAG\\nEMIT CHANGES;\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates a table from HASHSTREAM which count the tweets per hashtag in a 10 min window (TUMBLING)\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/HashTable.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "c583a9e7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100   846  100   603  100   243   1518    612 --:--:-- --:--:-- --:--:--  2125\n",
      "[\n",
      "   {\n",
      "      \"@type\" : \"currentStatus\",\n",
      "      \"commandId\" : \"table/`COUNT_USER_VERIFIED`/create\",\n",
      "      \"commandSequenceNumber\" : 12,\n",
      "      \"commandStatus\" : {\n",
      "         \"message\" : \"Created query with ID CTAS_COUNT_USER_VERIFIED_11\",\n",
      "         \"queryId\" : \"CTAS_COUNT_USER_VERIFIED_11\",\n",
      "         \"status\" : \"SUCCESS\"\n",
      "      },\n",
      "      \"statementText\" : \"CREATE TABLE COUNT_USER_VERIFIED WITH (KAFKA_TOPIC='COUNT_USER_VERIFIED', PARTITIONS=1, REPLICAS=1) AS SELECT\\n  RAWTWEETS.USER->SCREENNAME USER_SCREENNAME,\\n  COUNT(*) COUNT_TWEETS\\nFROM RAWTWEETS RAWTWEETS\\nWINDOW TUMBLING ( SIZE 5 MINUTES ) \\nWHERE RAWTWEETS.USER->VERIFIED\\nGROUP BY RAWTWEETS.USER->SCREENNAME\\nEMIT CHANGES;\",\n",
      "      \"warnings\" : []\n",
      "   }\n",
      "]\n"
     ]
    }
   ],
   "source": [
    "# It creates a table from RAWTWEETS which count the tweets per verified account.\n",
    "!curl -X POST  -H \"Content-Type: application/json; charset=utf-8\" -d @ksql/VerTable.json http://ksqldb-server:8088/ksql | json_pp"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}