def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,
        expand_action_callback=expand_action, raise_on_exception=True, 
        **kwargs):
    actions = map(expand_action_callback, actions)

    # if raise on error is set, we need to collect errors per chunk before raising them
    errors = []

    while True:
        chunk = islice(actions, chunk_size)

        # raise on exception means we might need to iterate on chunk twice
        if not raise_on_exception:
            chunk = list(chunk)

        bulk_actions = []
        for action, data in chunk:
            bulk_actions.append(action)
            if data is not None:
                bulk_actions.append(data)

        if not bulk_actions:
            return

        try:
            # send the actual request
            resp = client.bulk(bulk_actions, **kwargs)
        except TransportError as e:
            # default behavior - just propagate exception
            if raise_on_exception:
                raise e

            # if we are not propagating, mark all actions in current chunk as failed
            err_message = str(e)
            exc_errors = []
            for action, data in chunk:
                info = {"error": err_message, "status": e.status_code, "exception": e, "data": data}
                op_type, action = action.popitem()
                info.update(action)
                exc_errors.append({op_type: info})

            # emulate standard behavior for failed actions
            if raise_on_error:
                raise BulkIndexError('%i document(s) failed to index.' % len(exc_errors), exc_errors)
            else:
                for err in exc_errors:
                    yield False, err
                continue

        # go through request-reponse pairs and detect failures
        for op_type, item in map(methodcaller('popitem'), resp['items']):
            ok = 200 <= item.get('status', 500) < 300
            if not ok and raise_on_error:
                errors.append({op_type: item})

            if not errors:
                # if we are not just recording all errors to be able to raise
                # them all at once, yield items individually
                yield ok, {op_type: item}

        if errors:
            raise BulkIndexError('%i document(s) failed to index.' % len(errors), errors)
Streaming bulk consumes actions from the iterable passed in and yields
dataset/ETHPy150Open hvandenb/splunk-elasticsearch/search-elasticsearch/bin/elasticsearch/helpers/__init__.py/streaming_bulk