x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameIndexer.java
private Stream<IndexRequest> processBucketsToIndexRequests(CompositeAggregation agg) {
        final DataFrameTransformConfig transformConfig = getConfig();
        String indexName = transformConfig.getDestination();

        return pivot.extractResults(agg, getStats()).map(document -> {
            XContentBuilder builder;
            try {
                builder = jsonBuilder();
                builder.map(document);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }

            IndexRequest request = new IndexRequest(indexName).source(builder);
            return request;
        });
    }
/*
     * Parses the result and creates a stream of indexable documents
     *
     * Implementation decisions:
     *
     * Extraction uses generic maps as intermediate exchange format in order to hook in ingest pipelines/processors
     * in later versions, see {@link IngestDocument).
     */
Parses the result and creates a stream of indexable documents
Implementation decisions.

Extraction uses generic maps as intermediate exchange format in order to hook in ingest pipelines/processors
in later versions, see {@link IngestDocument).
Parses the result and creates a stream of indexable documents Implementation decisions . Extraction uses generic maps as intermediate exchange format in order to hook in ingest pipelines / processors in later versions see { @link IngestDocument ) .
private Stream<IndexRequest> processBucketsToIndexRequests(CompositeAggregation agg) {
        final DataFrameTransformConfig transformConfig = getConfig();
        String indexName = transformConfig.getDestination();
        return pivot.extractResults(agg, getStats()).map(document -> {
            XContentBuilder builder;
            try {
                builder = jsonBuilder();
                builder.map(document);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
            IndexRequest request = new IndexRequest(indexName).source(builder);
            return request;
        });
    }
private Stream < IndexRequest > processBucketsToIndexRequests ( CompositeAggregation agg ) { final DataFrameTransformConfig transformConfig = getConfig ( ) ; String indexName = transformConfig . getDestination ( ) ; return pivot . extractResults ( agg , getStats ( ) ) . map ( document -> { XContentBuilder builder ; try { builder = jsonBuilder ( ) ; builder . map ( document ) ; } catch ( IOException e ) { throw new UncheckedIOException ( e ) ; } IndexRequest request = new IndexRequest ( indexName ) . source ( builder ) ; return request ; } ) ; }
Parses the result and creates a stream of indexable documents
 Implementation decisions:
Parses the result and creates a stream of indexable documents Implementation decisions :