{"cells":[{"metadata":{"_uuid":"006380ad6101e0a7a92e54f806d5c118f5a2e372"},"cell_type":"markdown","source":"# 1. Quick start: read csv and flatten json fields + smart dump\n\nHi! This notebook is a derivative of https://www.kaggle.com/ogrellier/create-extracted-json-fields-dataset. I also tried to use [this kernel by julian3833](https://www.kaggle.com/julian3833/1-quick-start-read-csv-and-flatten-json-fields), but failed to execute `json_normalise` on a dask DataFrame. It extends the original code by using `dask.DataFrame`, see the docs [here](https://docs.dask.org/en/latest/dataframe.html). This allows to process data **with pandas-like interface in parallel threads and in chunks**. This allows to run faster and to work around the RAM limit.\n\n# Main goals\n1. **Process dataset, that can not fit into memory.**\n2. **Use dask toolkit that allows to scale data processing to a cluster instead of a single core.**\n3. **Store pre-processed flat data**\n\nThe output is stored in gziped csv file to reduce the file size. "},{"metadata":{"trusted":true,"_uuid":"f7a1ddb199f9a3dde2bd79e9f801cdb0113795a2"},"cell_type":"code","source":"import os\nimport json\nimport numpy as np\nimport pandas as pd\nfrom pandas.io.json import json_normalize\nimport pyarrow as pa\n\nimport dask\nimport dask.dataframe as dd\n\n# Set up a logger to dump messages to both log file and notebook\nimport logging as logging\ndef ini_log(filename):\n    logger = logging.getLogger(__name__)\n    logger.setLevel(logging.DEBUG)\n    \n    handlers = [logging.StreamHandler(None), logging.FileHandler(filename, 'a')]\n    \n    fmt=logging.Formatter('%(asctime)-15s: %(levelname)s  %(message)s')\n    for h in handlers:\n        h.setFormatter(fmt)\n        logger.addHandler(h)\n    return logger\n        \nlog = ini_log('out.log')\n#log.basicConfig(filename='out.log',level=log.DEBUG, format='%(asctime)-15s: %(levelname)s  %(message)s')\n\nimport gc\ngc.enable()","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"90c75750bfffc327985462432da39b4ac7c7eedf"},"cell_type":"markdown","source":"The original functions (from the aforementioned kernel)  with updates to run with `dask`"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"_kg_hide-input":true,"scrolled":true},"cell_type":"code","source":"def_num = np.nan\ndef_str = 'NaN'\n\ndef get_keys_for_field(field=None):\n    the_dict = {\n        'device': [\n            'browser', 'object',\n            'deviceCategory',\n            ('isMobile', False, bool),\n            'operatingSystem'\n        ],\n        'geoNetwork': [\n            'city',\n            'continent',\n            'country',\n            'metro',\n            'networkDomain',\n            'region',\n            'subContinent'\n        ],\n        'totals': [\n            ('pageviews', 0, np.int16),\n            ('hits', def_num, np.int16),\n            ('bounces', 0, np.int8),\n            ('newVisits', 0, np.int16),\n            ('transactionRevenue', 0, np.int64),\n            ('visits', -1, np.int16),\n            ('timeOnSite', -1, np.int32),\n            ('sessionQualityDim', -1, np.int8),\n        ],\n        'trafficSource': [\n            'adContent',\n            #'adwordsClickInfo',\n            'campaign',\n            ('isTrueDirect', False, bool),\n            #'keyword', #can not be saved in train (utf-8 symbols left)\n            'medium',\n            'referralPath',\n            'source'\n        ],\n    }\n    return the_dict[field]\n\n\ndef convert_to_dict(x):\n    #print(x, type(x))\n    return eval(x.replace('false', 'False')\n                .replace('true', 'True')\n                .replace('null', 'np.nan'))\n\ndef develop_json_fields(fin, json_fields=['totals'], bsize=1e8, cols_2drop=[]):\n    df = dd.read_csv(fin, blocksize=bsize, \n                 #converters={column: json.loads for column in JSON_COLUMNS},\n                 dtype={'fullVisitorId': 'str', # Important!!\n                        #usecols=lambda c: c not in cols_2drop,\n                            'date': 'str',\n                            **{c: 'str' for c in json_fields}\n                           },\n                     parse_dates=['date'],)#.head(10000, 100)\n    \n    df = df.drop(cols_2drop, axis=1)\n    \n    # Get the keys\n    for json_field in json_fields:\n        log.info('Doing Field {}'.format(json_field))\n        # Get json field keys to create columns\n        the_keys = get_keys_for_field(json_field)\n        # Replace the string by a dict\n        log.info('Transform string to dict')        \n        df[json_field] = df[json_field].apply(lambda x: convert_to_dict(x), meta=('','object'))\n        \n        log.info('{} converted to dict'.format(json_field))\n        #display(df.head())\n        for k in the_keys:\n            if isinstance(k, str):\n                t_ = def_str\n                k_ = k\n            else:\n                t_ = k[1]\n                k_ = k[0]\n            df[json_field + '_' + k_] = df[json_field].to_bag().pluck(k_, default=t_).to_dataframe().iloc[:,0]\n            if not isinstance(k, str) and len(k)>2:\n                df[json_field + '_' + k_] = df[json_field + '_' + k_].astype(k[2])\n            \n        del df[json_field]\n        gc.collect()\n        log.info('{} fields extracted'.format(json_field))\n    return df\n\nprint(os.listdir(\"../input\"))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"b468cfd85c0d9aa2feb809667b617d78a5b2c5cf","_kg_hide-output":true},"cell_type":"code","source":"!head ../input/train_v2.csv","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"90ef503c237aa3cc21a7ffc0bad0e5e4a3168134"},"cell_type":"markdown","source":"## Let's load the original data with pre-processing"},{"metadata":{"trusted":true,"_uuid":"0a969de0b49460ed5310a21f0f4ab016da6b447a","_kg_hide-input":true},"cell_type":"code","source":"JSON_COLUMNS = ['device', 'geoNetwork', 'totals', 'trafficSource']\nDROP_COLUMNS = ['customDimensions', 'hits', 'socialEngagementType']\n\ndef measure_memory(df, name):\n    size_df = df.memory_usage(deep=True)\n    log.info('{} size: {:.2f} MB'.format(name, size_df.sum().compute()/ 1024**2))\n    \ndef read_parse_store(fin, label='XXX', bsize=1e9):\n    log.debug('Start with {}'.format(label))\n    df_  = develop_json_fields(fin,  bsize=bsize, json_fields=JSON_COLUMNS, cols_2drop=DROP_COLUMNS)\n    \n    #some stats\n    measure_memory(df_, label)\n    log.info('Number of partitions in {}: {}'.format(label, df_.npartitions))\n    \n    #visualize a few rows\n    display(df_.head())\n    \n    #reduce var size\n    df_['visitNumber'] = df_['visitNumber'].astype(np.uint16)\n\n    #read the whole dataset into pd.DataFrame in memory and store into a single file\n    #otherwise dask.DataFrame would be stored into multiple files- 1 per partition\n    df_.compute().to_csv(\"{}-flat.csv.gz\".format(label), index=False , compression='gzip')","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"8148cfc802eac3c211b13a8b1db6dce0c3001a5e"},"cell_type":"markdown","source":"Process training data"},{"metadata":{"trusted":true,"_uuid":"648667084eb212c622580283302a84959b54dc3f"},"cell_type":"code","source":"%%time\nread_parse_store('../input/train_v2.csv', 'train')","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"4c035f18955926718b5e38bc7177c5bd29861d72"},"cell_type":"markdown","source":"Process test data"},{"metadata":{"trusted":true,"_uuid":"9bad553719e2fe31d36cb18620decf75c0cc2fda"},"cell_type":"code","source":"%%time\nread_parse_store('../input/test_v2.csv', 'test')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"06e210fc76416e284032a8f954e4d3bbca21b22e"},"cell_type":"code","source":"!ls -l","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"812409fd6ba717cf629d01766c1f834900bc6205"},"cell_type":"code","source":"","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.6.6","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"}},"nbformat":4,"nbformat_minor":1}