{"action":"created","issue":{"url":"https://api.github.com/repos/dask/dask/issues/1066","repository_url":"https://api.github.com/repos/dask/dask","labels_url":"https://api.github.com/repos/dask/dask/issues/1066/labels{/name}","comments_url":"https://api.github.com/repos/dask/dask/issues/1066/comments","events_url":"https://api.github.com/repos/dask/dask/issues/1066/events","html_url":"https://github.com/dask/dask/issues/1066","id":144155184,"number":1066,"title":"dask.bag: allow printing result to stdout","user":{"login":"Jeffrey04","id":46977,"avatar_url":"https://avatars.githubusercontent.com/u/46977?v=3","gravatar_id":"","url":"https://api.github.com/users/Jeffrey04","html_url":"https://github.com/Jeffrey04","followers_url":"https://api.github.com/users/Jeffrey04/followers","following_url":"https://api.github.com/users/Jeffrey04/following{/other_user}","gists_url":"https://api.github.com/users/Jeffrey04/gists{/gist_id}","starred_url":"https://api.github.com/users/Jeffrey04/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Jeffrey04/subscriptions","organizations_url":"https://api.github.com/users/Jeffrey04/orgs","repos_url":"https://api.github.com/users/Jeffrey04/repos","events_url":"https://api.github.com/users/Jeffrey04/events{/privacy}","received_events_url":"https://api.github.com/users/Jeffrey04/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2016-03-29T04:45:29Z","updated_at":"2016-03-29T21:08:50Z","closed_at":null,"body":"Suppose my use case is this, read input from stdin, process it, then output to stdout (so that it can be piped into another process). However this cannot be done easily as of now. I have cheated with `.map_partitions(lambda partition: [print(item) for item in partition])`, but due to the highly concurrent nature of dask it broke `print()` as it is not atomic (getting incomplete lines sometimes).\r\n\r\nI then hacked the `dask.multiprocessing.get` to make the manager a global option\r\n\r\n```\r\n    manager = _globals['manager'] or multiprocessing.Manager()\r\n```\r\n\r\nand then use `dask.set_options()` to replace the manager, so that I can write a printer to print from a queue instead of letting all N processes to print concurrently, as follows\r\n\r\n```\r\ndef output_printer(queue):\r\n    while True:\r\n        msg = queue.get()\r\n\r\n        if msg == '_DONE':\r\n            break\r\n        else:\r\n            print(json.dumps(msg))\r\n\r\ndef queue_send(partition, print_queue=None):\r\n    for item in partition:\r\n        print_queue.put(item)\r\n\r\nmanager = Manager()\r\ndask.set_options(manager=manager)\r\nprint_queue = manager.Queue()\r\nprinter = Process(target=output_printer, args=(print_queue, ))\r\nprinter.daemon = True\r\nprinter.start()\r\n\r\nbag = db.from_sequence(sys.stdin, 500) \\\r\n  .map(json.loads) \\\r\n  .map_partitions(process_the_jsonlines) \\\r\n  .map_partitions(queue_send, print_queue=queue) \\\r\n  .compute()\r\n\r\nprint_queue.put('_DONE')\r\nprinter.join()\r\n```\r\n\r\nHowever, is there a better way to do this? It would be nice to be able to print the result right after it is computed (`to_stdout()` ?), instead of waiting until the whole collection is done processing. I don't mind if it is not in the right order."},"comment":{"url":"https://api.github.com/repos/dask/dask/issues/comments/203106004","html_url":"https://github.com/dask/dask/issues/1066#issuecomment-203106004","issue_url":"https://api.github.com/repos/dask/dask/issues/1066","id":203106004,"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-03-29T21:08:50Z","updated_at":"2016-03-29T21:08:50Z","body":"The *best* way would probably be to fork off a process to do the actual printing with one end of a queue, and have the printing task put the results in the queue to be printed. This should limit the effects of synchronization, which using a lock causes. Assuming the process doing the printing can keep up with the workers creating the data this shouldn't flood your RAM either.\r\n\r\nA cheap option though is just to use a manager to create a lock. The following works for me:\r\n\r\n```python\r\nfrom multiprocessing import Manager\r\nimport dask.bag as db\r\n\r\nmanager = Manager()\r\nlock = manager.Lock()\r\ndata = db.range(1000, npartitions=10)\r\n\r\ndef to_stdout(data, lock=lock):\r\n    lock.acquire()\r\n    for row in data:\r\n        print(row)\r\n    lock.release()\r\n\r\ndata.map_partitions(to_stdout).compute()\r\n```\r\n\r\nSince grabbing a lock is mildly expensive, we do it once for each chunk rather than once for each record. If this is fast enough for you, then this is the solution I'd recommend."}}