{"action":"created","issue":{"url":"https://api.github.com/repos/pandas-dev/pandas/issues/22401","repository_url":"https://api.github.com/repos/pandas-dev/pandas","labels_url":"https://api.github.com/repos/pandas-dev/pandas/issues/22401/labels{/name}","comments_url":"https://api.github.com/repos/pandas-dev/pandas/issues/22401/comments","events_url":"https://api.github.com/repos/pandas-dev/pandas/issues/22401/events","html_url":"https://github.com/pandas-dev/pandas/issues/22401","id":351695783,"node_id":"MDU6SXNzdWUzNTE2OTU3ODM=","number":22401,"title":"Memory issues with `multiprocessing` and GroupBy on multiindex DataFrame","user":{"login":"ewengillies","id":5664063,"node_id":"MDQ6VXNlcjU2NjQwNjM=","avatar_url":"https://avatars3.githubusercontent.com/u/5664063?v=4","gravatar_id":"","url":"https://api.github.com/users/ewengillies","html_url":"https://github.com/ewengillies","followers_url":"https://api.github.com/users/ewengillies/followers","following_url":"https://api.github.com/users/ewengillies/following{/other_user}","gists_url":"https://api.github.com/users/ewengillies/gists{/gist_id}","starred_url":"https://api.github.com/users/ewengillies/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ewengillies/subscriptions","organizations_url":"https://api.github.com/users/ewengillies/orgs","repos_url":"https://api.github.com/users/ewengillies/repos","events_url":"https://api.github.com/users/ewengillies/events{/privacy}","received_events_url":"https://api.github.com/users/ewengillies/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-08-17T18:29:04Z","updated_at":"2018-08-17T20:08:17Z","closed_at":"2018-08-17T20:08:17Z","author_association":"NONE","body":"#### Code Sample, a copy-pastable example if possible\r\n\r\n```python\r\nimport pandas as pd\r\nimport multiprocessing as mp\r\nimport numpy as np\r\n%load_ext memory_profiler\r\n```\r\n\r\n\r\n```python\r\ndef working_apply_function(x):\r\n    \"\"\"\r\n    Working version of apply function\r\n    \"\"\"\r\n    var_a = x[\"var_a\"].values\r\n    var_b = x[\"var_b\"].values\r\n    # Hit index is accessed as a column\r\n    hit_idx = x[\"hit_index\"].values\r\n    # Normally some more complex function of var_a, var_b, and hit_idx\r\n    interesting_hits = hit_idx[[np.argmax(var_a), np.argmin(var_b)]]\r\n    return interesting_hits\r\n\r\ndef working_multiprocess(data_frame, n_proc=None):\r\n    \"\"\"\r\n    Get the hit index of interesting hits in a given event group, working version\r\n    \"\"\"\r\n    # Create a copy of the data with the columns that are needed\r\n    grp_data = pd.DataFrame(data_frame[[\"var_a\", \"var_b\"]])\r\n    # Add the hit index as a column.  This index that forms the basis of the output\r\n    grp_data.reset_index(level=\"hit_index\", drop=False, inplace=True)\r\n    # Group the data by the major index\r\n    grp_data = grp_data.groupby(\"event_index\")\r\n    # Find the interesting hit indexes\r\n    interesting_hit_idxs = None\r\n    # Use sequential mode if only one cpu requested\r\n    if n_proc is None:\r\n        interesting_hit_idxs = grp_data.apply(working_apply_function)\r\n        interesting_hit_idxs = interesting_hit_idxs.values\r\n    # Open up a pool of CPUs otherwise\r\n    else:\r\n        # Open the pool\r\n        with mp.Pool(n_proc) as pool:\r\n            interesting_hit_idxs = pool.map(working_apply_function,\r\n                                            [grp for _, grp in grp_data])\r\n    # Return the correct indexes\r\n    return np.sort(np.concatenate(interesting_hit_idxs))\r\n\r\ndef leaky_apply_function(x):\r\n    \"\"\"\r\n    Leaky version of apply_function where indexes are passed\r\n    \"\"\"\r\n    var_a = x[\"var_a\"].values\r\n    var_b = x[\"var_b\"].values\r\n    # Hit index is accessed as an index\r\n    hit_idx = x.index.get_level_values(\"hit_index\")\r\n    # Normally some more complex function of var_a, var_b, and hit_idx\r\n    interesting_hits = hit_idx[[np.argmax(var_a), np.argmin(var_b)]]\r\n    return interesting_hits\r\n\r\ndef leaky_multiprocess(data_frame, n_proc=None):\r\n    \"\"\"\r\n    Get the hit index of interesting hits in a given event group, leaky version\r\n    \"\"\"\r\n    # Group by event, but pass in the hit_index as an index\r\n    grp_data = data_frame[[\"var_a\", \"var_b\"]].groupby(\"event_index\")\r\n    # Find the interesting hits\r\n    interesting_hit_idxs = None\r\n    # Use sequential mode if only one cpu requested\r\n    if n_proc is None:\r\n        interesting_hit_idxs = grp_data.apply(leaky_apply_function)\r\n        interesting_hit_idxs = interesting_hit_idxs.dropna().values\r\n        # Return None if there are no hit ids\r\n        if interesting_hit_idxs.shape[0] == 0:\r\n            return None\r\n    # Open up a pool of CPUs otherwise\r\n    else:\r\n        # Open the pool\r\n        with mp.Pool(n_proc) as pool:\r\n            interesting_hit_idxs = pool.map(leaky_apply_function, [grp for _, grp in grp_data])\r\n        # Return None if there are none\r\n        if not interesting_hit_idxs:\r\n            return None\r\n    # Return the correct indexes\r\n    return np.sort(np.concatenate(interesting_hit_idxs))\r\n```\r\n\r\n\r\n```python\r\n# Define a large number of columns\r\nn_hits = int(1e6)\r\nhits_per_event = int(4e3)\r\nn_events = n_hits//hits_per_event\r\nprint(n_events, n_hits, hits_per_event)\r\n# Define the data frame\r\ntest_data_frame = pd.DataFrame(data={\"var_a\": np.random.random(size=n_hits),\r\n                                     \"var_b\": np.random.random(size=n_hits)})\r\n# Set the indexes as columns\r\ntest_data_frame[\"event_index\"] = np.concatenate([np.repeat(evt, hits_per_event) for evt in range(n_events)])\r\ntest_data_frame[\"hit_index\"] = np.arange(n_hits)\r\n# Set them as indexes in place and drop them\r\ntest_data_frame.set_index([\"event_index\", \"hit_index\"], drop=True, inplace=True)\r\ntest_data_frame.info()\r\n```\r\n\r\n    250 1000000 4000\r\n    <class 'pandas.core.frame.DataFrame'>\r\n    MultiIndex: 1000000 entries, (0, 0) to (249, 999999)\r\n    Data columns (total 2 columns):\r\n    var_a    1000000 non-null float64\r\n    var_b    1000000 non-null float64\r\n    dtypes: float64(2)\r\n    memory usage: 28.6 MB\r\n\r\n```python\r\ntest_data_frame.head()\r\n```\r\n\r\n\r\n<table border=\"1\" class=\"dataframe\">\r\n  <thead>\r\n    <tr style=\"text-align: right;\">\r\n      <th></th>\r\n      <th></th>\r\n      <th>var_a</th>\r\n      <th>var_b</th>\r\n    </tr>\r\n    <tr>\r\n      <th>event_index</th>\r\n      <th>hit_index</th>\r\n      <th></th>\r\n      <th></th>\r\n    </tr>\r\n  </thead>\r\n  <tbody>\r\n    <tr>\r\n      <th rowspan=\"5\" valign=\"top\">0</th>\r\n      <th>0</th>\r\n      <td>0.564295</td>\r\n      <td>0.019199</td>\r\n    </tr>\r\n    <tr>\r\n      <th>1</th>\r\n      <td>0.836201</td>\r\n      <td>0.586676</td>\r\n    </tr>\r\n    <tr>\r\n      <th>2</th>\r\n      <td>0.660377</td>\r\n      <td>0.218596</td>\r\n    </tr>\r\n    <tr>\r\n      <th>3</th>\r\n      <td>0.887197</td>\r\n      <td>0.324807</td>\r\n    </tr>\r\n    <tr>\r\n      <th>4</th>\r\n      <td>0.454777</td>\r\n      <td>0.479819</td>\r\n    </tr>\r\n  </tbody>\r\n</table>\r\n</div>\r\n\r\n\r\n\r\n\r\n```python\r\n# In sequencial mode, both work fine\r\n%memit interest_idx_work_seq = working_multiprocess(test_data_frame, n_proc=None)\r\n```\r\n\r\n    > peak memory: 233.51 MiB, increment: 61.47 MiB\r\n\r\n\r\n\r\n```python\r\n%memit interest_idx_leak_seq = leaky_multiprocess(test_data_frame, n_proc=None)\r\n```\r\n\r\n    > peak memory: 205.43 MiB, increment: 17.32 MiB\r\n\r\n\r\n\r\n```python\r\nnp.testing.assert_allclose(interest_idx_work_seq, interest_idx_leak_seq)\r\n```\r\n\r\n\r\n```python\r\n# In parallel mode, the working version does not hurt the memory\r\n%memit interest_idx_work_multiproc = working_multiprocess(test_data_frame, n_proc=1)\r\n```\r\n\r\n    > peak memory: 246.51 MiB, increment: 41.06 MiB\r\n\r\n\r\n\r\n```python\r\n# In parallel mode, the leaky version kills the memory, even with only one process!\r\n%memit interest_idx_leak_multiproc = leaky_multiprocess(test_data_frame, n_proc=1)\r\n```\r\n```\r\n    peak memory: 1509.04 MiB, increment: 1262.78 MiB\r\n```\r\n\r\n#### Problem description\r\n\r\nLet me start by saying I am new to `pandas`, but have long since used `numpy`.  Pandas is great, and thank you for all your effort.  This issue may be less of a bug and more of a consequence of trying to parallelize pandas.  If so, I hope it is useful for others.\r\n\r\nIt seems that passing an indexed object from a multiindexed data frame that is grouped on one index eats up memory when `mulitprocessing` is used.  This may be because the entire remaining index is passed to the group.  This is normally returned as a view of this index, but I speculate that `multiprocessing` copies the entire index.\r\n\r\nThis suspicion lead me to the working version.  This issue is similar,  #19630, and provided the hint I needed to get to the solution.\r\n\r\nAbove, I've recreated a very basic version of my use-case.  This use-case has a multi-index dataframe with an `event` index that groups columns together, and a `hit` index that uniquely enumerates all rows.  Really, my `apply_func` function is much more complicated, but the idea is that it determines `hit` indexes of interesting hits, as defined by values in other columns.\r\n\r\nOnce this data is passed to `multiprocessing`, the memory usage skyrockets.  I tried to profile it using `%memit`, as above.  My actual usage can be seen here, the bump is from `leaky_multiprocess` with `n_proc=1`, peaks around 10GB.  The other calls are flat on this scale\r\n\r\n![image](https://user-images.githubusercontent.com/5664063/44282182-236a7f00-a252-11e8-93c7-1e45a7fa76e2.png)\r\n\r\n#### Expected Output\r\n\r\nSimilar memory performance in all four cases posted. Instead the last case eats around 8GB.\r\n\r\n#### Output of ``pd.show_versions()``\r\n```\r\n<details>INSTALLED VERSIONS\r\n------------------\r\ncommit: None\r\npython: 3.6.5.final.0\r\npython-bits: 64\r\nOS: Linux    \r\nOS-release: 4.17.8-041708-generic\r\nmachine: x86_64\r\nprocessor: x86_64\r\nbyteorder: little\r\nLC_ALL: None   \r\nLANG: en_GB.UTF-8\r\nLOCALE: en_GB.UTF-8\r\n                \r\npandas: 0.23.4\r\npytest: 3.7.1 \r\npip: 10.0.1  \r\nsetuptools: 39.2.0\r\nCython: None  \r\nnumpy: 1.14.3\r\nscipy: 1.1.0\r\npyarrow: None   \r\nxarray: None\r\nIPython: 6.4.0\r\nsphinx: None   \r\npatsy: None     \r\ndateutil: 2.7.3\r\npytz: 2018.4  \r\nblosc: None \r\nbottleneck: None\r\ntables: 3.4.3    \r\nnumexpr: 2.6.5  \r\nfeather: None          \r\nmatplotlib: 2.2.2\r\nopenpyxl: None\r\nxlrd: None\r\nxlwt: None\r\nxlsxwriter: None\r\nlxml: None\r\nbs4: None\r\nhtml5lib: 1.0.1\r\nsqlalchemy: None\r\npymysql: None\r\npsycopg2: None\r\njinja2: 2.10\r\ns3fs: None\r\nfastparquet: None\r\npandas_gbq: None\r\npandas_datareader: None\r\n```"},"comment":{"url":"https://api.github.com/repos/pandas-dev/pandas/issues/comments/413975262","html_url":"https://github.com/pandas-dev/pandas/issues/22401#issuecomment-413975262","issue_url":"https://api.github.com/repos/pandas-dev/pandas/issues/22401","id":413975262,"node_id":"MDEyOklzc3VlQ29tbWVudDQxMzk3NTI2Mg==","user":{"login":"jreback","id":953992,"node_id":"MDQ6VXNlcjk1Mzk5Mg==","avatar_url":"https://avatars2.githubusercontent.com/u/953992?v=4","gravatar_id":"","url":"https://api.github.com/users/jreback","html_url":"https://github.com/jreback","followers_url":"https://api.github.com/users/jreback/followers","following_url":"https://api.github.com/users/jreback/following{/other_user}","gists_url":"https://api.github.com/users/jreback/gists{/gist_id}","starred_url":"https://api.github.com/users/jreback/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jreback/subscriptions","organizations_url":"https://api.github.com/users/jreback/orgs","repos_url":"https://api.github.com/users/jreback/repos","events_url":"https://api.github.com/users/jreback/events{/privacy}","received_events_url":"https://api.github.com/users/jreback/received_events","type":"User","site_admin":false},"created_at":"2018-08-17T20:08:17Z","updated_at":"2018-08-17T20:08:17Z","author_association":"CONTRIBUTOR","body":"your example is not minimal \r\n\r\nyou are very unlikely to get any help here\r\nand are better off asking on SO\r\n\r\nfurther raw multi processing has many issues\r\n\r\nyou are much better off using dask"}}
{"actor":{"display_login":"jreback"}}