01tweet_json_to_parquet.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import orjson as json \n",
    "import gzip\n",
    "import os\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_path = '/data_volume/streaming-data-UTC/'\n",
    "output_path = '/data_volume/pub/antivax/raw_tweets_parquets/' "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "start_date = '2021-02-01'\n",
    "end_date = '2021-11-01'\n",
    "start_date = datetime.strptime(start_date, '%Y-%m-%d')\n",
    "end_date = datetime.strptime(end_date, '%Y-%m-%d')  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['streaming_data--2021-02-01.json.gz',\n",
       " 'streaming_data--2021-02-02.json.gz',\n",
       " 'streaming_data--2021-02-03.json.gz',\n",
       " 'streaming_data--2021-02-04.json.gz',\n",
       " 'streaming_data--2021-02-05.json.gz',\n",
       " 'streaming_data--2021-02-06.json.gz',\n",
       " 'streaming_data--2021-02-07.json.gz',\n",
       " 'streaming_data--2021-02-08.json.gz',\n",
       " 'streaming_data--2021-02-09.json.gz',\n",
       " 'streaming_data--2021-02-10.json.gz',\n",
       " 'streaming_data--2021-02-11.json.gz',\n",
       " 'streaming_data--2021-02-12.json.gz',\n",
       " 'streaming_data--2021-02-13.json.gz',\n",
       " 'streaming_data--2021-02-14.json.gz',\n",
       " 'streaming_data--2021-02-15.json.gz',\n",
       " 'streaming_data--2021-02-16.json.gz',\n",
       " 'streaming_data--2021-02-17.json.gz',\n",
       " 'streaming_data--2021-02-18.json.gz',\n",
       " 'streaming_data--2021-02-19.json.gz',\n",
       " 'streaming_data--2021-02-20.json.gz',\n",
       " 'streaming_data--2021-02-21.json.gz',\n",
       " 'streaming_data--2021-02-22.json.gz',\n",
       " 'streaming_data--2021-02-23.json.gz',\n",
       " 'streaming_data--2021-02-24.json.gz',\n",
       " 'streaming_data--2021-02-25.json.gz',\n",
       " 'streaming_data--2021-02-26.json.gz',\n",
       " 'streaming_data--2021-02-27.json.gz',\n",
       " 'streaming_data--2021-02-28.json.gz',\n",
       " 'streaming_data--2021-03-01.json.gz',\n",
       " 'streaming_data--2021-03-02.json.gz',\n",
       " 'streaming_data--2021-03-03.json.gz',\n",
       " 'streaming_data--2021-03-04.json.gz',\n",
       " 'streaming_data--2021-03-05.json.gz',\n",
       " 'streaming_data--2021-03-06.json.gz',\n",
       " 'streaming_data--2021-03-07.json.gz',\n",
       " 'streaming_data--2021-03-08.json.gz',\n",
       " 'streaming_data--2021-03-09.json.gz',\n",
       " 'streaming_data--2021-03-10.json.gz',\n",
       " 'streaming_data--2021-03-11.json.gz',\n",
       " 'streaming_data--2021-03-12.json.gz',\n",
       " 'streaming_data--2021-03-13.json.gz',\n",
       " 'streaming_data--2021-03-14.json.gz',\n",
       " 'streaming_data--2021-03-15.json.gz',\n",
       " 'streaming_data--2021-03-16.json.gz',\n",
       " 'streaming_data--2021-03-17.json.gz',\n",
       " 'streaming_data--2021-03-18.json.gz',\n",
       " 'streaming_data--2021-03-19.json.gz',\n",
       " 'streaming_data--2021-03-20.json.gz',\n",
       " 'streaming_data--2021-03-21.json.gz',\n",
       " 'streaming_data--2021-03-22.json.gz',\n",
       " 'streaming_data--2021-03-23.json.gz',\n",
       " 'streaming_data--2021-03-24.json.gz',\n",
       " 'streaming_data--2021-03-25.json.gz',\n",
       " 'streaming_data--2021-03-26.json.gz',\n",
       " 'streaming_data--2021-03-27.json.gz',\n",
       " 'streaming_data--2021-03-28.json.gz',\n",
       " 'streaming_data--2021-03-29.json.gz',\n",
       " 'streaming_data--2021-03-30.json.gz',\n",
       " 'streaming_data--2021-03-31.json.gz',\n",
       " 'streaming_data--2021-04-01.json.gz',\n",
       " 'streaming_data--2021-04-02.json.gz',\n",
       " 'streaming_data--2021-04-03.json.gz',\n",
       " 'streaming_data--2021-04-04.json.gz',\n",
       " 'streaming_data--2021-04-05.json.gz',\n",
       " 'streaming_data--2021-04-06.json.gz',\n",
       " 'streaming_data--2021-04-07.json.gz',\n",
       " 'streaming_data--2021-04-08.json.gz',\n",
       " 'streaming_data--2021-04-09.json.gz',\n",
       " 'streaming_data--2021-04-10.json.gz',\n",
       " 'streaming_data--2021-04-11.json.gz',\n",
       " 'streaming_data--2021-04-12.json.gz',\n",
       " 'streaming_data--2021-04-13.json.gz',\n",
       " 'streaming_data--2021-04-14.json.gz',\n",
       " 'streaming_data--2021-04-15.json.gz',\n",
       " 'streaming_data--2021-04-16.json.gz',\n",
       " 'streaming_data--2021-04-17.json.gz',\n",
       " 'streaming_data--2021-04-18.json.gz',\n",
       " 'streaming_data--2021-04-19.json.gz',\n",
       " 'streaming_data--2021-04-20.json.gz',\n",
       " 'streaming_data--2021-04-21.json.gz',\n",
       " 'streaming_data--2021-04-22.json.gz',\n",
       " 'streaming_data--2021-04-23.json.gz',\n",
       " 'streaming_data--2021-04-24.json.gz',\n",
       " 'streaming_data--2021-04-25.json.gz',\n",
       " 'streaming_data--2021-04-26.json.gz',\n",
       " 'streaming_data--2021-04-27.json.gz',\n",
       " 'streaming_data--2021-04-28.json.gz',\n",
       " 'streaming_data--2021-04-29.json.gz',\n",
       " 'streaming_data--2021-04-30.json.gz',\n",
       " 'streaming_data--2021-05-01.json.gz',\n",
       " 'streaming_data--2021-05-02.json.gz',\n",
       " 'streaming_data--2021-05-03.json.gz',\n",
       " 'streaming_data--2021-05-04.json.gz',\n",
       " 'streaming_data--2021-05-05.json.gz',\n",
       " 'streaming_data--2021-05-06.json.gz',\n",
       " 'streaming_data--2021-05-07.json.gz',\n",
       " 'streaming_data--2021-05-08.json.gz',\n",
       " 'streaming_data--2021-05-09.json.gz',\n",
       " 'streaming_data--2021-05-10.json.gz',\n",
       " 'streaming_data--2021-05-11.json.gz',\n",
       " 'streaming_data--2021-05-12.json.gz',\n",
       " 'streaming_data--2021-05-13.json.gz',\n",
       " 'streaming_data--2021-05-14.json.gz',\n",
       " 'streaming_data--2021-05-15.json.gz',\n",
       " 'streaming_data--2021-05-16.json.gz',\n",
       " 'streaming_data--2021-05-17.json.gz',\n",
       " 'streaming_data--2021-05-18.json.gz',\n",
       " 'streaming_data--2021-05-19.json.gz',\n",
       " 'streaming_data--2021-05-20.json.gz',\n",
       " 'streaming_data--2021-05-21.json.gz',\n",
       " 'streaming_data--2021-05-22.json.gz',\n",
       " 'streaming_data--2021-05-23.json.gz',\n",
       " 'streaming_data--2021-05-24.json.gz',\n",
       " 'streaming_data--2021-05-25.json.gz',\n",
       " 'streaming_data--2021-05-26.json.gz',\n",
       " 'streaming_data--2021-05-27.json.gz',\n",
       " 'streaming_data--2021-05-28.json.gz',\n",
       " 'streaming_data--2021-05-29.json.gz',\n",
       " 'streaming_data--2021-05-30.json.gz',\n",
       " 'streaming_data--2021-05-31.json.gz',\n",
       " 'streaming_data--2021-06-01.json.gz',\n",
       " 'streaming_data--2021-06-02.json.gz',\n",
       " 'streaming_data--2021-06-03.json.gz',\n",
       " 'streaming_data--2021-06-04.json.gz',\n",
       " 'streaming_data--2021-06-05.json.gz',\n",
       " 'streaming_data--2021-06-06.json.gz',\n",
       " 'streaming_data--2021-06-07.json.gz',\n",
       " 'streaming_data--2021-06-08.json.gz',\n",
       " 'streaming_data--2021-06-09.json.gz',\n",
       " 'streaming_data--2021-06-10.json.gz',\n",
       " 'streaming_data--2021-06-11.json.gz',\n",
       " 'streaming_data--2021-06-12.json.gz',\n",
       " 'streaming_data--2021-06-13.json.gz',\n",
       " 'streaming_data--2021-06-14.json.gz',\n",
       " 'streaming_data--2021-06-15.json.gz',\n",
       " 'streaming_data--2021-06-16.json.gz',\n",
       " 'streaming_data--2021-06-17.json.gz',\n",
       " 'streaming_data--2021-06-18.json.gz',\n",
       " 'streaming_data--2021-06-19.json.gz',\n",
       " 'streaming_data--2021-06-20.json.gz',\n",
       " 'streaming_data--2021-06-21.json.gz',\n",
       " 'streaming_data--2021-06-22.json.gz',\n",
       " 'streaming_data--2021-06-23.json.gz',\n",
       " 'streaming_data--2021-06-24.json.gz',\n",
       " 'streaming_data--2021-06-25.json.gz',\n",
       " 'streaming_data--2021-06-26.json.gz',\n",
       " 'streaming_data--2021-06-27.json.gz',\n",
       " 'streaming_data--2021-06-28.json.gz',\n",
       " 'streaming_data--2021-06-29.json.gz',\n",
       " 'streaming_data--2021-06-30.json.gz',\n",
       " 'streaming_data--2021-07-01.json.gz',\n",
       " 'streaming_data--2021-07-02.json.gz',\n",
       " 'streaming_data--2021-07-03.json.gz',\n",
       " 'streaming_data--2021-07-04.json.gz',\n",
       " 'streaming_data--2021-07-05.json.gz',\n",
       " 'streaming_data--2021-07-06.json.gz',\n",
       " 'streaming_data--2021-07-07.json.gz',\n",
       " 'streaming_data--2021-07-08.json.gz',\n",
       " 'streaming_data--2021-07-09.json.gz',\n",
       " 'streaming_data--2021-07-10.json.gz',\n",
       " 'streaming_data--2021-07-11.json.gz',\n",
       " 'streaming_data--2021-07-12.json.gz',\n",
       " 'streaming_data--2021-07-13.json.gz',\n",
       " 'streaming_data--2021-07-14.json.gz',\n",
       " 'streaming_data--2021-07-15.json.gz',\n",
       " 'streaming_data--2021-07-16.json.gz',\n",
       " 'streaming_data--2021-07-17.json.gz',\n",
       " 'streaming_data--2021-07-18.json.gz',\n",
       " 'streaming_data--2021-07-19.json.gz',\n",
       " 'streaming_data--2021-07-20.json.gz',\n",
       " 'streaming_data--2021-07-21.json.gz',\n",
       " 'streaming_data--2021-07-22.json.gz',\n",
       " 'streaming_data--2021-07-23.json.gz',\n",
       " 'streaming_data--2021-07-24.json.gz',\n",
       " 'streaming_data--2021-07-25.json.gz',\n",
       " 'streaming_data--2021-07-26.json.gz',\n",
       " 'streaming_data--2021-07-27.json.gz',\n",
       " 'streaming_data--2021-07-28.json.gz',\n",
       " 'streaming_data--2021-07-29.json.gz',\n",
       " 'streaming_data--2021-07-30.json.gz',\n",
       " 'streaming_data--2021-07-31.json.gz',\n",
       " 'streaming_data--2021-08-01.json.gz',\n",
       " 'streaming_data--2021-08-02.json.gz',\n",
       " 'streaming_data--2021-08-03.json.gz',\n",
       " 'streaming_data--2021-08-04.json.gz',\n",
       " 'streaming_data--2021-08-05.json.gz',\n",
       " 'streaming_data--2021-08-06.json.gz',\n",
       " 'streaming_data--2021-08-07.json.gz',\n",
       " 'streaming_data--2021-08-08.json.gz',\n",
       " 'streaming_data--2021-08-09.json.gz',\n",
       " 'streaming_data--2021-08-10.json.gz',\n",
       " 'streaming_data--2021-08-11.json.gz',\n",
       " 'streaming_data--2021-08-12.json.gz',\n",
       " 'streaming_data--2021-08-13.json.gz',\n",
       " 'streaming_data--2021-08-14.json.gz',\n",
       " 'streaming_data--2021-08-15.json.gz',\n",
       " 'streaming_data--2021-08-16.json.gz',\n",
       " 'streaming_data--2021-08-17.json.gz',\n",
       " 'streaming_data--2021-08-18.json.gz',\n",
       " 'streaming_data--2021-08-19.json.gz',\n",
       " 'streaming_data--2021-08-20.json.gz',\n",
       " 'streaming_data--2021-08-21.json.gz',\n",
       " 'streaming_data--2021-08-22.json.gz',\n",
       " 'streaming_data--2021-08-23.json.gz',\n",
       " 'streaming_data--2021-08-24.json.gz',\n",
       " 'streaming_data--2021-08-25.json.gz',\n",
       " 'streaming_data--2021-08-26.json.gz',\n",
       " 'streaming_data--2021-08-27.json.gz',\n",
       " 'streaming_data--2021-08-28.json.gz',\n",
       " 'streaming_data--2021-08-29.json.gz',\n",
       " 'streaming_data--2021-08-30.json.gz',\n",
       " 'streaming_data--2021-08-31.json.gz',\n",
       " 'streaming_data--2021-09-01.json.gz',\n",
       " 'streaming_data--2021-09-02.json.gz',\n",
       " 'streaming_data--2021-09-03.json.gz',\n",
       " 'streaming_data--2021-09-04.json.gz',\n",
       " 'streaming_data--2021-09-05.json.gz',\n",
       " 'streaming_data--2021-09-06.json.gz',\n",
       " 'streaming_data--2021-09-07.json.gz',\n",
       " 'streaming_data--2021-09-08.json.gz',\n",
       " 'streaming_data--2021-09-09.json.gz',\n",
       " 'streaming_data--2021-09-10.json.gz',\n",
       " 'streaming_data--2021-09-11.json.gz',\n",
       " 'streaming_data--2021-09-12.json.gz',\n",
       " 'streaming_data--2021-09-13.json.gz',\n",
       " 'streaming_data--2021-09-14.json.gz',\n",
       " 'streaming_data--2021-09-15.json.gz',\n",
       " 'streaming_data--2021-09-16.json.gz',\n",
       " 'streaming_data--2021-09-17.json.gz',\n",
       " 'streaming_data--2021-09-18.json.gz',\n",
       " 'streaming_data--2021-09-19.json.gz',\n",
       " 'streaming_data--2021-09-20.json.gz',\n",
       " 'streaming_data--2021-09-21.json.gz',\n",
       " 'streaming_data--2021-09-22.json.gz',\n",
       " 'streaming_data--2021-09-23.json.gz',\n",
       " 'streaming_data--2021-09-24.json.gz',\n",
       " 'streaming_data--2021-09-25.json.gz',\n",
       " 'streaming_data--2021-09-26.json.gz',\n",
       " 'streaming_data--2021-09-27.json.gz',\n",
       " 'streaming_data--2021-09-28.json.gz',\n",
       " 'streaming_data--2021-09-29.json.gz',\n",
       " 'streaming_data--2021-09-30.json.gz',\n",
       " 'streaming_data--2021-10-01.json.gz',\n",
       " 'streaming_data--2021-10-02.json.gz',\n",
       " 'streaming_data--2021-10-03.json.gz',\n",
       " 'streaming_data--2021-10-04.json.gz',\n",
       " 'streaming_data--2021-10-05.json.gz',\n",
       " 'streaming_data--2021-10-06.json.gz',\n",
       " 'streaming_data--2021-10-07.json.gz',\n",
       " 'streaming_data--2021-10-08.json.gz',\n",
       " 'streaming_data--2021-10-09.json.gz',\n",
       " 'streaming_data--2021-10-10.json.gz',\n",
       " 'streaming_data--2021-10-11.json.gz',\n",
       " 'streaming_data--2021-10-12.json.gz',\n",
       " 'streaming_data--2021-10-13.json.gz',\n",
       " 'streaming_data--2021-10-14.json.gz',\n",
       " 'streaming_data--2021-10-15.json.gz',\n",
       " 'streaming_data--2021-10-16.json.gz',\n",
       " 'streaming_data--2021-10-17.json.gz',\n",
       " 'streaming_data--2021-10-18.json.gz',\n",
       " 'streaming_data--2021-10-19.json.gz',\n",
       " 'streaming_data--2021-10-20.json.gz',\n",
       " 'streaming_data--2021-10-21.json.gz',\n",
       " 'streaming_data--2021-10-22.json.gz',\n",
       " 'streaming_data--2021-10-23.json.gz',\n",
       " 'streaming_data--2021-10-24.json.gz',\n",
       " 'streaming_data--2021-10-25.json.gz',\n",
       " 'streaming_data--2021-10-26.json.gz',\n",
       " 'streaming_data--2021-10-27.json.gz',\n",
       " 'streaming_data--2021-10-28.json.gz',\n",
       " 'streaming_data--2021-10-29.json.gz',\n",
       " 'streaming_data--2021-10-30.json.gz',\n",
       " 'streaming_data--2021-10-31.json.gz',\n",
       " 'streaming_data--2021-11-01.json.gz']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#get list of files to process. exclude files already processed. \n",
    "dates = [file[-18:-8] for file in os.listdir(input_path) if '.json.gz' in file]\n",
    "skip_dates = set([file[-18:-8] for file in os.listdir(output_path) if '.parquet' in file])\n",
    "dates = [date for date in dates if date not in skip_dates]\n",
    "files = ['streaming_data--'+date+'.json.gz' for date in dates]\n",
    "\n",
    "#filer and sort by date\n",
    "files = [file for file in files if start_date <= datetime.strptime(file[16:26], '%Y-%m-%d') <= end_date] \n",
    "files.sort()\n",
    "files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_full_text(tweet):\n",
    "    if 'retweeted_status' in tweet and 'extended_tweet' in tweet['retweeted_status']:\n",
    "        text = tweet['retweeted_status']['extended_tweet']['full_text']\n",
    "    elif 'retweeted_status' in tweet:\n",
    "        text = tweet['retweeted_status']['text']\n",
    "    elif 'extended_tweet' in tweet:\n",
    "        text = tweet['extended_tweet']['full_text']\n",
    "    else:\n",
    "        text = tweet['text']\n",
    "    return text\n",
    "\n",
    "tweets_cols = ['tweet_id', 'text', 'tweet_time', 'user_id', 'lang', 'retweeted_tweet_id']\n",
    "def get_tweet_tuple(tweet): \n",
    "    text = extract_full_text(tweet)\n",
    "    retweeted_tweet_id = int(tweet['retweeted_status']['id']) if 'retweeted_status' in tweet else -1\n",
    "    tweet_tuple = (tweet['id'], \n",
    "             text, \n",
    "             tweet['created_at'],\n",
    "             tweet['user']['id'],\n",
    "             tweet['lang'],\n",
    "             retweeted_tweet_id)\n",
    "    return tweet_tuple"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 3h 32min 45s, sys: 7min 21s, total: 3h 40min 6s\n",
      "Wall time: 3h 40min 10s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "#extract data from json\n",
    "for i, file in enumerate(files):\n",
    "    with gzip.open(input_path+file, 'rb') as zipfile: #open raw data file\n",
    "        tweets = list()\n",
    "        for j, line in enumerate(zipfile):\n",
    "            tweet = json.loads(line) #read one tweet object from json\n",
    "            if 'limit' in tweet: continue #skips rate limit error objects\n",
    "            tweets.append(get_tweet_tuple(tweet))\n",
    "    tweets = pd.DataFrame(tweets, columns = tweets_cols)\n",
    "    tweets['tweet_time'] = pd.to_datetime(tweets['tweet_time'], format = '%a %b %d %H:%M:%S %z %Y'\n",
    "                                         ).dt.tz_localize(None)\n",
    "    tweets.to_parquet(output_path+'tweets'+file[-18:-8]+'.parquet')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}