{"cells":[{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"# Inspired by notebook from Aguiar\n# Func: Extract all features, flatten json columns\n# with multiprocessing module\nimport pandas as pd\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nimport json\nimport gc\nimport sys\nimport math\nfrom pandas.io.json import json_normalize\nfrom datetime import datetime\nimport os\nimport time\nfrom datetime import datetime\nfrom ast import literal_eval\nimport multiprocessing\nimport glob\npd.set_option('display.max_columns', 500)\npd.set_option('display.max_columns', 500)\npd.set_option('display.max_rows', 500)\npd.set_option('display.max_columns', 500)\npd.set_option('display.width', 1000)\n\n# Any results you write to the current directory are saved as output.","execution_count":null,"outputs":[]},{"metadata":{"_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","trusted":true},"cell_type":"code","source":"# Define the dataframe processing module\nJSON_COLUMNS = ['device', 'geoNetwork', 'totals', 'trafficSource']\ndef work_on_one_reader(df,index,flag):\n    df.reset_index(drop = True,inplace = True)\n    for column in JSON_COLUMNS:\n        column_as_df = json_normalize(df[column])\n        column_as_df.columns = [f\"{column}.{subcolumn}\" for subcolumn in column_as_df.columns]\n        df = df.drop(column, axis=1).merge(column_as_df, right_index=True, left_index=True)\n\n        # Normalize customDimensions\n    df['customDimensions']=df['customDimensions'].apply(literal_eval)\n    df['customDimensions']=df['customDimensions'].str[0]\n    df['customDimensions']=df['customDimensions'].apply(lambda x: {'index':np.NaN,'value':np.NaN} if pd.isnull(x) else x)\n\n    column_as_df = json_normalize(df['customDimensions'])\n    column_as_df.columns = [f\"customDimensions_{subcolumn}\" for subcolumn in column_as_df.columns]\n    df = df.drop('customDimensions', axis=1).merge(column_as_df, right_index=True, left_index=True)\n\n    # ===============================================\n#     print(\"---> working on hits\")\n    feat = 'hits'\n    df[feat]=df[feat].apply(literal_eval)\n    df[feat]=df[feat].str[0]\n    df[feat]=df[feat].apply(lambda x: {'index':np.NaN} if pd.isnull(x) else x)\n    column_as_df = json_normalize(df[feat])\n    column_as_df.columns = [f\"hits_{subcolumn}\" for subcolumn in column_as_df.columns]\n    df = df.drop(feat, axis=1).merge(column_as_df, right_index=True, left_index=True)\n\n    # ===============================================\n#     print(\"---> working on hits_promotion\")\n    feat = 'hits_promotion'\n#     df[feat]=df[feat].apply(literal_eval)\n    df[feat]=df[feat].str[0]\n    df[feat]=df[feat].apply(lambda x: {'index':np.NaN} if pd.isnull(x) else x)\n    column_as_df = json_normalize(df[feat])\n    column_as_df.columns = [f\"hits_promotion_{subcolumn}\" for subcolumn in column_as_df.columns]\n    df = df.drop(feat, axis=1).merge(column_as_df, right_index=True, left_index=True)\n\n\n    # ===============================================\n#     print(\"---> working on hits_product\")\n    feat = 'hits_product'\n#     df[feat]=df[feat].apply(literal_eval)\n    df[feat]=df[feat].str[0]\n    df[feat]=df[feat].apply(lambda x: {'index':np.NaN} if pd.isnull(x) else x)\n    column_as_df = json_normalize(df[feat])\n    column_as_df.columns = [f\"hits_product_{subcolumn}\" for subcolumn in column_as_df.columns]\n    df = df.drop(feat, axis=1).merge(column_as_df, right_index=True, left_index=True)\n\n    bracket_col = ['hits_customDimensions','hits_customMetrics','hits_customVariables','hits_experiment',\n               'hits_publisher_infos','hits_product_customDimensions','hits_product_customMetrics']\n    for col in bracket_col:\n        df[col] = df[col].str[0]\n    out_name = \"../input/out.{0}.{1}.csv\".format(flag,index)\n    df.to_csv(out_name,index=False)\n    ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"8839c96281ee9a3a0c5eae6d3da006e164e34a3f"},"cell_type":"code","source":"def process_input_csv(csv_path,chunksize=5000,flag=None):\n    time_beg = datetime.now()\n    dfs = pd.read_csv(csv_path, sep=',',\n            converters={column: json.loads for column in JSON_COLUMNS}, \n            dtype={'fullVisitorId': 'str'}, # Important!!\n            chunksize=chunksize)\n    jobs = []\n    for index,df in enumerate(dfs):\n        print(\"--> job {} started\".format(index))\n        p = multiprocessing.Process(target=work_on_one_reader,args=(df,index,flag,))\n        p.start()\n        jobs.append(p)\n    for index,job in enumerate(jobs):\n        job.join()\n    \n    # read in all processed csv file and concat together\n    path = \"../input\"\n    all_file = glob.glob(path+\"/out.{}.*.csv\".format(flag))\n    list_ = []\n    new_df = pd.DataFrame()\n    for file in all_file:\n        df = pd.read_csv(file,low_memory=False)\n        list_.append(df)\n    new_df = pd.concat(list_)\n    time_end = datetime.now()\n    print(\"----> Total time spent is {}\".format(time_end - time_beg))\n    return new_df","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"d957effbcd5e9164d788401b9064a140857964f1"},"cell_type":"code","source":"test_df = process_input_csv(\"../input/test_v2.csv\",chunksize=5000,flag=\"test\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true,"_uuid":"41a8730ba4f9dc578fae964bfba390a560641f48"},"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}