#NEWLINE# def parallel_bulk( client , actions , thread_count = 4 , chunk_size = 500 , max_chunk_bytes = ( ( 100 * 1024 ) * 1024 ) , expand_action_callback = expand_action , **kwargs ) : #NEWLINE# #INDENT# "\n    Parallel version of the bulk helper run in multiple threads at once.\n\n    :arg client: instance of :class:`~elasticsearch.Elasticsearch` to use\n    :arg actions: iterator containing the actions\n    :arg thread_count: size of the threadpool to use for the bulk requests\n    :arg chunk_size: number of docs in one chunk sent to es (default: 500)\n    :arg max_chunk_bytes: the maximum size of the request in bytes (default: 100MB)\n    :arg raise_on_error: raise ``BulkIndexError`` containing errors (as `.errors`)\n        from the execution of the last chunk when some occur. By default we raise.\n    :arg raise_on_exception: if ``False`` then don't propagate exceptions from\n        call to ``bulk`` and just report the items that failed as failed.\n    :arg expand_action_callback: callback executed on each action passed in,\n        should return a tuple containing the action line and the data line\n        (`None` if data line should be omitted).\n    " #NEWLINE# from multiprocessing.dummy import Pool #NEWLINE# actions = map ( expand_action_callback , actions ) #NEWLINE# pool = Pool ( thread_count ) #NEWLINE# try : #NEWLINE# #INDENT# for result in pool . imap ( ( lambda chunk : list ( _process_bulk_chunk ( client , chunk , ** kwargs ) ) ) , _chunk_actions ( actions , chunk_size , max_chunk_bytes , client . transport . serializer ) ) : #NEWLINE# #INDENT# for item in result : #NEWLINE# #INDENT# ( yield item ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# finally : #NEWLINE# #INDENT# pool . close ( ) #NEWLINE# pool . join ( )