HDFS_DB_Init.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyhive import hive\n",
    "import pyarrow as pa\n",
    "from config import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs = pa.hdfs.connect(\n",
    "    host=hdfs_host, \n",
    "    port=hdfs_port, \n",
    "    user=hive_username, \n",
    "    kerb_ticket=None, \n",
    "    driver='libhdfs', \n",
    "    extra_conf=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#USE WITH CAUTION\n",
    "\n",
    "#DELETE and recreate the data and checkpoint directories in hdfs \n",
    "\n",
    "\n",
    "\n",
    "def reset_dir(hdfs_path):\n",
    "    if fs.exists(hdfs_path):\n",
    "        fs.delete(hdfs_path,recursive=True)\n",
    "        print(\"directory deleted: \" + hdfs_path)\n",
    "    else:\n",
    "        print(\"no such dir: \" + hdfs_path)\n",
    "\n",
    "    fs.mkdir(hdfs_path + '/sparkcheckpoint')    \n",
    "    print(\"directory created: \" + fs.ls(hdfs_path)[0])\n",
    "    fs.chmod(hdfs_path + '/sparkcheckpoint', 0o777)   \n",
    "    fs.chmod(hdfs_path, 0o777)    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_queries(queries):\n",
    "    \"\"\"\n",
    "    queries -- must be list of strings, e.g. [\"select from...\", \"insert into...\" ]\n",
    "    \"\"\"\n",
    "    with hive_cnx.cursor() as cursor:   \n",
    "        for q in queries:\n",
    "            cursor.execute(q)    \n",
    "        if cursor.poll().hasResultSet:\n",
    "            return cursor.fetchall()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reset_dir(hdfs_archive_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run_queries()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create  HIVE Database"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hive_cnx = hive.Connection(\n",
    "  host = hdfs_host, \n",
    "  port = hive_port, \n",
    "  username = hive_username,\n",
    "  password = hive_password,\n",
    "  auth = hive_mode)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# q = f\"drop database if exists {hive_database} cascade \"\n",
    "\n",
    "# with hive_cnx.cursor() as cursor:\n",
    "#     cursor.execute(q)\n",
    "\n",
    "#reset_dir(f\"{hdfs_hive_warehouse}/{hive_database}.db\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run_queries([f\"create database if not exists {hive_database}\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hive_cnx = hive.Connection(\n",
    "  host = hdfs_host, \n",
    "  port = hive_port, \n",
    "  username = hive_username,\n",
    "  password = hive_password,\n",
    "  auth = hive_mode,\n",
    "  database=hive_database)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create tweets table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table = \"tweets\"\n",
    "colNames = tweet_keys.copy()\n",
    "colNames += [\"Sentiment\",\"n_words\"]\n",
    "\n",
    "colTypes = [t().simpleString() for t in tweet_types]\n",
    "colTypes += ['int','int']\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "colCreate = ', '.join([n + ' ' + t for n,t in zip(colNames, colTypes)])\n",
    "colCreate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reset_dir(hdfs_hive_tweets)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q =  f\"CREATE EXTERNAL TABLE IF NOT EXISTS {table} ({colCreate}) \\\n",
    "partitioned by (key string)\\n \\\n",
    "STORED AS PARQUET \\\n",
    "LOCATION  '{hdfs_hive_tweets}'\"\n",
    "\n",
    "# f\"LOAD DATA INPATH '{hdfs_hive_staging}/*.parquet' \\\n",
    "# INTO TABLE tweets \\\n",
    "# partition by key \"\n",
    "print(q)\n",
    "\n",
    "#TBLPROPERTIES ('avro.schema.url'='hdfs://$tempdir/avroSchema/$tbl.avsc') ;\" "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run_queries([q])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Users Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "table = \"users\"\n",
    "staging = \"users\" # must actually be \"users_staging\", but the aggregation strategy does not work (see below) \n",
    "colNames = ['last_tweet_at'] + user_keys[:-1]\n",
    "colTypes = [t().simpleString() for t in [StringType] + user_types][:-1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "colCreate = ', '.join([n + ' ' + t for n,t in zip(colNames, colTypes)])\n",
    "colCreate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reset_dir(hdfs_hive_users)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "reset_dir(hdfs_hive_users_staging)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1 =  [f\"CREATE EXTERNAL TABLE IF NOT EXISTS {table} ({colCreate}) \\\n",
    "partitioned by ({partitionCol} string) \\\n",
    "STORED AS PARQUET \\\n",
    "LOCATION '{hdfs_hive_users}'\"]\n",
    "\n",
    "# f\"LOAD DATA INPATH '{hdfs_hive_staging}/*.parquet' \\\n",
    "# INTO TABLE tweets \\\n",
    "# partition by key \"\n",
    "print(q1)\n",
    "\n",
    "#TBLPROPERTIES ('avro.schema.url'='hdfs://$tempdir/avroSchema/$tbl.avsc') ;\" "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1 +=  [f\"CREATE TABLE IF NOT EXISTS {staging} ({colCreate}) \\\n",
    "partitioned by ({partitionCol} string) \\\n",
    "STORED AS PARQUET \\\n",
    "LOCATION '{hdfs_hive_users_staging}'\"]\n",
    "\n",
    "# f\"LOAD DATA INPATH '{hdfs_hive_staging}/*.parquet' \\\n",
    "# INTO TABLE tweets \\\n",
    "# partition by key \"\n",
    "print(q1)\n",
    "\n",
    "#TBLPROPERTIES ('avro.schema.url'='hdfs://$tempdir/avroSchema/$tbl.avsc') ;\" "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run_queries(q1)"
   ]
  }
 ],
 "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.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}