{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Reading .json files using DASK\n\nI'm creating this notebook to show you guys how i managed to speed up the read process of the .json files, i came across a few methods on other people's notebooks and it was taking around 13 minutes to read all 14k train files, a few months ago i came across a notebook that was loading many files at once using dask, so i tried to do something similar here, using dask the time to read was reduced to 4 minutes (on my 8 core machine), the upside of using this is that you will be able to run the same code in a personal laptop or a several-cores server, just adjusting the number of workers.\n\nDask can process cpu tasks in parallel, so the same method can be adapted to process any cpu intensive tasks that you might be using.","metadata":{}},{"cell_type":"code","source":"import re\nimport json\nimport dask\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom dask.distributed import Client, wait, LocalCluster","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:39:48.262615Z","iopub.execute_input":"2021-06-17T21:39:48.263001Z","iopub.status.idle":"2021-06-17T21:39:48.691980Z","shell.execute_reply.started":"2021-06-17T21:39:48.262920Z","shell.execute_reply":"2021-06-17T21:39:48.690987Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"BASE = '../input/coleridgeinitiative-show-us-the-data'\n\nPARAMS = {\n    'base_path': {BASE},\n    'labels_file': f'{BASE}/train.csv',\n    'train_folder': f'{BASE}/train',\n#     'labels_file': f'{BASE}/sample_submission.csv',\n#     'publications_path': f'{BASE}/test',\n}","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:39:48.693477Z","iopub.execute_input":"2021-06-17T21:39:48.693812Z","iopub.status.idle":"2021-06-17T21:39:48.698641Z","shell.execute_reply.started":"2021-06-17T21:39:48.693778Z","shell.execute_reply":"2021-06-17T21:39:48.697384Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def clean_text(input_words):\n    return re.sub('[^A-Za-z0-9\\[\\]]+', ' ', str(input_words).lower()).strip()","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:39:48.700504Z","iopub.execute_input":"2021-06-17T21:39:48.700780Z","iopub.status.idle":"2021-06-17T21:39:48.719539Z","shell.execute_reply.started":"2021-06-17T21:39:48.700752Z","shell.execute_reply":"2021-06-17T21:39:48.718282Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_df = pd.read_csv(PARAMS['labels_file'])\ndisplay(train_df)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:39:48.720810Z","iopub.execute_input":"2021-06-17T21:39:48.721015Z","iopub.status.idle":"2021-06-17T21:39:48.919789Z","shell.execute_reply.started":"2021-06-17T21:39:48.720994Z","shell.execute_reply":"2021-06-17T21:39:48.918753Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_files = train_df['Id'].unique()\nprint('unique files: {}'.format(len(train_files)))","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:39:48.920789Z","iopub.execute_input":"2021-06-17T21:39:48.920981Z","iopub.status.idle":"2021-06-17T21:39:48.932137Z","shell.execute_reply.started":"2021-06-17T21:39:48.920960Z","shell.execute_reply":"2021-06-17T21:39:48.931117Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Without dask","metadata":{}},{"cell_type":"code","source":"%%time\nfutures = []\nfor file in tqdm(train_files):\n\n    file_dfs = []\n    section = {}\n    \n    with open(PARAMS['train_folder']+'/'+ file +'.json', 'r') as f:\n        paper = json.load(f)\n\n    dfs = []    \n    len_paper = len(paper)\n        \n    for section_index in range (0, len_paper):\n        section_sentences = paper[section_index].get('text')\n        \n        section['file'] = file\n        section['orig_sentence'] = section_sentences\n        \n        df = pd.DataFrame.from_dict(section, orient = 'index').T\n        dfs.append(df)\n        \n        file_df = pd.concat(dfs)\n               \n    file_dfs.append(file_df)\n    file_df = pd.concat(file_dfs)\n    \n    futures.append(file_df)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T21:40:29.251413Z","iopub.execute_input":"2021-06-17T21:40:29.251681Z","iopub.status.idle":"2021-06-17T21:55:37.868885Z","shell.execute_reply.started":"2021-06-17T21:40:29.251656Z","shell.execute_reply":"2021-06-17T21:55:37.868310Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"sentences_df = pd.concat(futures).reset_index(drop=True)\ndisplay(sentences_df)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:00:22.017821Z","iopub.execute_input":"2021-06-17T22:00:22.018187Z","iopub.status.idle":"2021-06-17T22:00:24.069358Z","shell.execute_reply.started":"2021-06-17T22:00:22.018146Z","shell.execute_reply":"2021-06-17T22:00:24.068130Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Start a local cluster on the machine\n\nworkers = 4 means that we are going to use all 4 cores to process","metadata":{}},{"cell_type":"code","source":"# set n_workers to number of cores\nclient = Client(n_workers=4, threads_per_worker=4)\nclient","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:00:30.920958Z","iopub.execute_input":"2021-06-17T22:00:30.921493Z","iopub.status.idle":"2021-06-17T22:00:32.736560Z","shell.execute_reply.started":"2021-06-17T22:00:30.921448Z","shell.execute_reply":"2021-06-17T22:00:32.735980Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def extract_text_from_pub(file):\n\n    file_dfs = []\n    section = {}\n    \n    with open(PARAMS['train_folder']+'/'+ file +'.json', 'r') as f:\n        paper = json.load(f)\n\n    dfs = []    \n    len_paper = len(paper)\n        \n    for section_index in range (0, len_paper):\n        section_sentences = paper[section_index].get('text')\n        \n        section['file'] = file\n        section['orig_sentence'] = section_sentences\n        \n        df = pd.DataFrame.from_dict(section, orient = 'index').T\n        dfs.append(df)\n        \n        file_df = pd.concat(dfs)\n               \n    file_dfs.append(file_df)\n    file_df = pd.concat(file_dfs)\n    \n    return file_df","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:00:45.796659Z","iopub.execute_input":"2021-06-17T22:00:45.796929Z","iopub.status.idle":"2021-06-17T22:00:45.805991Z","shell.execute_reply.started":"2021-06-17T22:00:45.796905Z","shell.execute_reply":"2021-06-17T22:00:45.804510Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"NOTE: tqdm bar won't work properly here, it will only show the time it took to send the parameters to dask, so it will reach 100% much faster than the cell is going to finish","metadata":{}},{"cell_type":"code","source":"%%time\n# you will notice that the kernel CPU usage are close to 400% while this is running\n\nfutures = [] # save the future since dask is lazy, otherwise nothing is executed.\nfor file in tqdm(train_files):\n    f = client.submit(extract_text_from_pub,file) # pass the function to be executed and the file id\n    futures.append(f)\n_ = wait(futures)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:00:48.472612Z","iopub.execute_input":"2021-06-17T22:00:48.472935Z","iopub.status.idle":"2021-06-17T22:08:30.670297Z","shell.execute_reply.started":"2021-06-17T22:00:48.472906Z","shell.execute_reply":"2021-06-17T22:08:30.669049Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"%%time\nparagraphs = []\nfor f in tqdm(futures):\n    p = f.result()\n    paragraphs.append(p)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:08:30.671732Z","iopub.execute_input":"2021-06-17T22:08:30.672000Z","iopub.status.idle":"2021-06-17T22:09:20.481526Z","shell.execute_reply.started":"2021-06-17T22:08:30.671973Z","shell.execute_reply":"2021-06-17T22:09:20.480453Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"sentences_df = pd.concat(paragraphs).reset_index(drop=True)\ndisplay(sentences_df)","metadata":{"execution":{"iopub.status.busy":"2021-06-17T22:09:20.483782Z","iopub.execute_input":"2021-06-17T22:09:20.484217Z","iopub.status.idle":"2021-06-17T22:09:22.564674Z","shell.execute_reply.started":"2021-06-17T22:09:20.484181Z","shell.execute_reply":"2021-06-17T22:09:22.563569Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\n\n\n\n\n\n\n\n\n\n\n\n\n\n","metadata":{},"execution_count":null,"outputs":[]}]}