<commit_before>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Credentials - keep this secret!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#  Cluster number, e.g. 10000\n",
    "cluster  = ''\n",
    "\n",
    "# Cluster username\n",
    "username = ''\n",
    "\n",
    "# Cluster password\n",
    "password = ''\n",
    "\n",
    "# file path in HDFS\n",
    "filepath = 'yourpath/yourfile.csv'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Add your custom code to read_csv_lines for processing your datafile    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "        \n",
    "def read_csv_lines(lines, is_first_chunk = False):\n",
    "    ''' returns a pandas dataframe '''\n",
    "    \n",
    "    if is_first_chunk:\n",
    "        # you will want to set the header here if your datafile has a header record\n",
    "        return pd.read_csv(lines, sep='|', header=None)\n",
    "    else:\n",
    "        return pd.read_csv(lines, sep='|', header=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "host = 'ehaasp-{0}-mastermanager.bi.services.bluemix.net'.format(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "utility method to create dataframe from chunks - don't change this"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def unionAll(df, *dfs):\n",
    "    if not dfs:\n",
    "        raise ValueError()\n",
    "    first = dfs[0]\n",
    "    return df.sql_ctx.createDataFrame(\n",
    "        df._sc.union([df.rdd for df in dfs]), first.schema\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Code to connect to BigInsights on Cloud via WebHDFS - don't change this"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "import numpy as np\n",
    "import sys\n",
    "import datetime\n",
    "\n",
    "if sys.version_info[0] < 3: \n",
    "    from StringIO import StringIO\n",
    "else:\n",
    "    from io import StringIO\n",
    "    \n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "print('SCRIPT START: {0}'.format(datetime.datetime.now()))\n",
    "\n",
    "chunk_size = 1000000 # Read in 1 Mb chunks\n",
    "\n",
    "url = \"https://{0}:8443/gateway/default/webhdfs/v1/{1}?op=OPEN\".format(host, filepath)\n",
    "\n",
    "# note SSL verification is been disabled\n",
    "r = requests.get(url, \n",
    "                 auth=(username, password), \n",
    "                 verify=False, \n",
    "                 allow_redirects=True, \n",
    "                 stream=True)\n",
    "\n",
    "df = None\n",
    "chunk_num = 1\n",
    "remainder = ''\n",
    "for chunk in r.iter_content(chunk_size):\n",
    "    \n",
    "    if chunk: # filter out keep-alive new chunks\n",
    "        \n",
    "        # Show progress by printing a dot   \n",
    "        sys.stdout.write('.')\n",
    "        sys.stdout.flush()\n",
    "\n",
    "        txt = remainder + chunk\n",
    "        if '\\n' in txt:\n",
    "            [lines, remainder] = txt.rsplit('\\n', 1)\n",
    "        else:\n",
    "            lines = txt\n",
    "\n",
    "        if chunk_num == 1:\n",
    "            # First chunk may need to be processed differently to the other chunks\n",
    "            # E.g. if it has a header\n",
    "            pdf = read_csv_lines(StringIO(lines), True)\n",
    "        else:\n",
    "            pdf = read_csv_lines(StringIO(lines), False)\n",
    "\n",
    "        if df == None:\n",
    "            df = sqlContext.createDataFrame(pdf)\n",
    "        else:\n",
    "            df = unionAll(df, sqlContext.createDataFrame(pdf))\n",
    "            \n",
    "        chunk_num = chunk_num + 1\n",
    "print '\\nRecord count: {0}'.format(df.count())\n",
    "\n",
    "print('SCRIPT END: {0}'.format(datetime.datetime.now()))\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Add your code here to work with the imported dataframe, df"
   ]
  },<commit_msg>Update WebHDFS_Example.ipynb<commit_after>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Credentials - keep this secret!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#  Cluster number, e.g. 10000\n",
    "cluster  = ''\n",
    "\n",
    "# Cluster username\n",
    "username = ''\n",
    "\n",
    "# Cluster password\n",
    "password = ''\n",
    "\n",
    "# file path in HDFS\n",
    "filepath = 'yourpath/yourfile.csv'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Add your custom code to read_csv_lines for processing your datafile    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "        \n",
    "def read_csv_lines(lines, is_first_chunk = False):\n",
    "    ''' returns a pandas dataframe '''\n",
    "    \n",
    "    if is_first_chunk:\n",
    "        # you will want to set the header here if your datafile has a header record\n",
    "        return pd.read_csv(lines, sep='|', header=None)\n",
    "    else:\n",
    "        return pd.read_csv(lines, sep='|', header=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "host = 'ehaasp-{0}-mastermanager.bi.services.bluemix.net'.format(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "utility method to create dataframe from chunks - don't change this"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def unionAll(df, *dfs):\n",
    "    if not dfs:\n",
    "        raise ValueError()\n",
    "    first = dfs[0]\n",
    "    return df.sql_ctx.createDataFrame(\n",
    "        df._sc.union([df.rdd for df in dfs]), first.schema\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Code to connect to BigInsights on Cloud via WebHDFS - don't change this"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "import numpy as np\n",
    "import sys\n",
    "import datetime\n",
    "\n",
    "if sys.version_info[0] < 3: \n",
    "    from StringIO import StringIO\n",
    "else:\n",
    "    from io import StringIO\n",
    "    \n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "print('SCRIPT START: {0}'.format(datetime.datetime.now()))\n",
    "\n",
    "chunk_size = 1000000 # Read in 1 Mb chunks\n",
    "\n",
    "url = \"https://{0}:8443/gateway/default/webhdfs/v1/{1}?op=OPEN\".format(host, filepath)\n",
    "\n",
    "# note SSL verification is been disabled\n",
    "r = requests.get(url, \n",
    "                 auth=(username, password), \n",
    "                 verify=False, \n",
    "                 allow_redirects=True, \n",
    "                 stream=True)\n",
    "\n",
    "df = None\n",
    "chunk_num = 1\n",
    "remainder = ''\n",
    "for chunk in r.iter_content(chunk_size):\n",
    "    \n",
    "    if chunk: # filter out keep-alive new chunks\n",
    "        \n",
    "        # Show progress by printing a dot   \n",
    "        sys.stdout.write('.')\n",
    "        sys.stdout.flush()\n",
    "\n",
    "        txt = remainder + chunk\n",
    "        if '\\n' in txt:\n",
    "            [lines, remainder] = txt.rsplit('\\n', 1)\n",
    "        else:\n",
    "            lines = txt\n",
    "\n",
    "        if chunk_num == 1:\n",
    "            # First chunk may need to be processed differently to the other chunks\n",
    "            # E.g. if it has a header\n",
    "            pdf = read_csv_lines(StringIO(lines), True)\n",
    "        else:\n",
    "            pdf = read_csv_lines(StringIO(lines), False)\n",
    "\n",
    "        if df == None:\n",
    "            df = sqlContext.createDataFrame(pdf)\n",
    "        else:\n",
    "            df = unionAll(df, sqlContext.createDataFrame(pdf))\n",
    "            \n",
    "        chunk_num = chunk_num + 1\n",
    "            \n",
    "print '\\nRecord count: {0}'.format(df.count())\n",
    "\n",
    "print('SCRIPT END: {0}'.format(datetime.datetime.now()))\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Add your code here to work with the imported dataframe, df"
   ]
  },
{'fraction_non_alphanumeric': 0.19417168197656, 'fraction_numerical': 0.007390983000739098, 'mean_word_length': 1.9117737473101752, 'pattern_counts': {'":': 96, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5365096', 'n_tokens_mistral': 3502, 'n_tokens_neox': 3408, 'n_words': 730}