{"action":"created","issue":{"url":"https://api.github.com/repos/dask/dask/issues/1814","repository_url":"https://api.github.com/repos/dask/dask","labels_url":"https://api.github.com/repos/dask/dask/issues/1814/labels{/name}","comments_url":"https://api.github.com/repos/dask/dask/issues/1814/comments","events_url":"https://api.github.com/repos/dask/dask/issues/1814/events","html_url":"https://github.com/dask/dask/issues/1814","id":191838120,"number":1814,"title":"Overcoming `dask.async.MemoryError` with merge","user":{"login":"evanbiederstedt","id":6968193,"avatar_url":"https://avatars.githubusercontent.com/u/6968193?v=3","gravatar_id":"","url":"https://api.github.com/users/evanbiederstedt","html_url":"https://github.com/evanbiederstedt","followers_url":"https://api.github.com/users/evanbiederstedt/followers","following_url":"https://api.github.com/users/evanbiederstedt/following{/other_user}","gists_url":"https://api.github.com/users/evanbiederstedt/gists{/gist_id}","starred_url":"https://api.github.com/users/evanbiederstedt/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/evanbiederstedt/subscriptions","organizations_url":"https://api.github.com/users/evanbiederstedt/orgs","repos_url":"https://api.github.com/users/evanbiederstedt/repos","events_url":"https://api.github.com/users/evanbiederstedt/events{/privacy}","received_events_url":"https://api.github.com/users/evanbiederstedt/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2016-11-27T02:27:26Z","updated_at":"2016-12-07T18:39:44Z","closed_at":"2016-12-07T18:39:44Z","body":"I begin with two dataframes, working with Python 3.5 on a system with 2 TB RAM and approx. 300 threads:\r\n\r\ndf1 = pd.read_cvs(\"first1.csv\")\r\ndf2 = pd.read_csv(\"second2.csv\")\r\n\r\nThe size of `first1.csv` is 139 MB, and the size of `second2.csv` is 118 MB. As for the contents of these dataframes:\r\n\r\n```\r\nprint(df1.shape) # outputs (4757076, 4)\r\ndf1.head \r\n\r\n    column1  begin    end    category\r\n0  class1  10001  10468    third\r\n1  class1  10469  11447     third\r\n2  class1  11505  11675     fourth\r\n3  class2  15265  15355   seventh\r\n4  class2  15798  15849   second\r\n```\r\n\r\nand \r\n\r\n```\r\nprint(df2.shape) # (428764, 45)\r\n   column1  begin    end     category\r\n0  class1  11524   11556    fourteenth\r\n1  class1  11541   ....\r\n2  class1  11549  ....\r\n3  class1  11565  ...\r\n4  class1  11596  ...\r\n```\r\n\r\nI try to run the following code\r\n\r\n```\r\nimport pandas as pd\r\nimport os\r\nimport dask.dataframe as dd\r\nfrom numpy import nan\r\n\r\nddf1 = dd.from_pandas(repeats, npartitions=1000)  # I was hoping more parititions would make this managable\r\nddf2 = dd.from_pandas(anno, npartitions=1000)\r\n\r\nmerged = dd.merge(ddf1, ddf2, on=\"column1\", how=\"outer\", suffixes=(\"\",\"_repeat\"))\r\nmerged.to_csv(\"out.*.test1.csv\") \r\n # without a globstring with an asterisk `*`, I get this error: `dask.async.FileNotFoundError: [Errno 2] No such file or directory: 'out.test1.csv/865.part'`\r\n```\r\n\r\nThis last line fails:\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"<stdin>\", line 1, in <module>\r\n  File \"/path/lib/python3.5/site-packages/dask/dataframe/core.py\", line 985, in to_csv\r\n    return to_csv(self, filename, **kwargs)\r\n  File \"/path/lib/python3.5/site-packages/dask/dataframe/io/csv.py\", line 305, in to_csv\r\n    compute(*values, get=get)\r\n  File \"/path/lib/python3.5/site-packages/dask/base.py\", line 178, in compute\r\n    results = get(dsk, keys, **kwargs)\r\n  File \"/path/lib/python3.5/site-packages/dask/threaded.py\", line 69, in get\r\n    **kwargs)\r\n  File \"/path/lib/python3.5/site-packages/dask/async.py\", line 502, in get_async\r\n    raise(remote_exception(res, tb))\r\ndask.async.MemoryError: \r\n\r\nTraceback\r\n---------\r\n  File \"/path/lib/python3.5/site-packages/dask/async.py\", line 268, in execute_task\r\n    result = _execute_task(task, data)\r\n  File \"/path/lib/python3.5/site-packages/dask/async.py\", line 249, in _execute_task\r\n    return func(*args2)\r\n  File \"/path/python3.5/site-packages/dask/dataframe/methods.py\", line 221, in merge\r\n    suffixes=suffixes, indicator=indicator)\r\n  File \"/path/lib/python3.5/site-packages/pandas/tools/merge.py\", line 59, in merge\r\n    return op.get_result()\r\n  File \"/path/lib/python3.5/site-packages/pandas/tools/merge.py\", line 517, in get_result\r\n    concat_axis=0, copy=self.copy)\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/internals.py\", line 4818, in concatenate_block_managers\r\n    placement=placement) for placement, join_units in concat_plan]\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/internals.py\", line 4818, in <listcomp>\r\n    placement=placement) for placement, join_units in concat_plan]\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/internals.py\", line 4915, in concatenate_join_units\r\n    for ju in join_units]\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/internals.py\", line 4915, in <listcomp>\r\n    for ju in join_units]\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/internals.py\", line 5215, in get_reindexed_values\r\n    fill_value=fill_value)\r\n  File \"/path/lib/python3.5/site-packages/pandas/core/algorithms.py\", line 1074, in take_nd\r\n    out = np.empty(out_shape, dtype=dtype)\r\n```\r\nSo, how could I accomplish this with dask? Again, I'm using 2 TB of RAM. I tried doing this with more threads, but it's the same problem: \r\n\r\n    merged.to_csv(\"out.*.test1.csv\").compute(num_workers=60)\r\n\r\nI get the exact same error, `dask.async.MemoryError:`. So, how would I overcome this? My understanding as that there should be a way via dask to partition this merged dataframe such that one saves to disk without holding the entire structure in memory. "},"comment":{"url":"https://api.github.com/repos/dask/dask/issues/comments/265533228","html_url":"https://github.com/dask/dask/issues/1814#issuecomment-265533228","issue_url":"https://api.github.com/repos/dask/dask/issues/1814","id":265533228,"user":{"login":"jcrist","id":2783717,"avatar_url":"https://avatars.githubusercontent.com/u/2783717?v=3","gravatar_id":"","url":"https://api.github.com/users/jcrist","html_url":"https://github.com/jcrist","followers_url":"https://api.github.com/users/jcrist/followers","following_url":"https://api.github.com/users/jcrist/following{/other_user}","gists_url":"https://api.github.com/users/jcrist/gists{/gist_id}","starred_url":"https://api.github.com/users/jcrist/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jcrist/subscriptions","organizations_url":"https://api.github.com/users/jcrist/orgs","repos_url":"https://api.github.com/users/jcrist/repos","events_url":"https://api.github.com/users/jcrist/events{/privacy}","received_events_url":"https://api.github.com/users/jcrist/received_events","type":"User","site_admin":false},"created_at":"2016-12-07T18:39:44Z","updated_at":"2016-12-07T18:39:44Z","body":"Hi Evan, I sent you a more thorough response via email. For your use case, you're creating a huge intermediate table with ~2e12 rows (which is the error happens). Instead of using `merge`, I'd solve this by writing a function to do the label applying (using one of the dataframes as a lookup-table), and just using pandas to load the data and apply function. This won't result in the huge intermediate result, and should be sufficient for the size of inputs you have.\r\n\r\nClosing for now, feel free to reopen if needed."}}
{"actor":{"display_login":"jcrist"}}