diff --git a/docs/track.rst b/docs/track.rst
index ec849cb1..4fcfefb1 100644
--- a/docs/track.rst
+++ b/docs/track.rst
@@ -308,9 +308,10 @@ bulk
 With the operation type ``bulk`` you can execute `bulk requests <http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html>`_. It supports the following properties:
 
 * ``bulk-size`` (mandatory): Defines the bulk size in number of documents.
+* ``ingest-percentage`` (optional, defaults to 100): A number between (0, 100] that defines how much of the document corpus will be bulk-indexed.
 * ``corpora`` (optional): A list of document corpus names that should be targeted by this bulk-index operation. Only needed if the ``corpora`` section contains more than one document corpus and you don't want to index all of them with this operation.
 * ``indices`` (optional): A list of index names that defines which indices should be used by this bulk-index operation. Rally will then only select the documents files that have a matching ``target-index`` specified.
-* ``batch-size`` (optional): Defines how many documents Rally will read at once. This is an expert setting and only meant to avoid accidental bottlenecks for very small bulk sizes (e.g. if you want to benchmark with a bulk-size of 1, you should set batch-size higher).
+* ``batch-size`` (optional): Defines how many documents Rally will read at once. This is an expert setting and only meant to avoid accidental bottlenecks for very small bulk sizes (e.g. if you want to benchmark with a bulk-size of 1, you should set ``batch-size`` higher).
 * ``pipeline`` (optional): Defines the name of an (existing) ingest pipeline that should be used (only supported from Elasticsearch 5.0).
 * ``conflicts`` (optional): Type of index conflicts to simulate. If not specified, no conflicts will be simulated. Valid values are: 'sequential' (A document id is replaced with a document id with a sequentially increasing id), 'random' (A document id is replaced with a document id with a random other id).
 
diff --git a/esrally/track/params.py b/esrally/track/params.py
index e9838253..48ef30eb 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -1,6 +1,7 @@
 import logging
 import random
 import time
+import math
 import types
 import inspect
 from enum import Enum
@@ -482,6 +483,14 @@ class BulkIndexParamSource(ParamSource):
         except ValueError:
             raise exceptions.InvalidSyntax("'batch-size' must be numeric")
 
+        try:
+            self.ingest_percentage = float(params.get("ingest-percentage", 100.0))
+            if self.ingest_percentage <= 0 or self.ingest_percentage > 100.0:
+                raise exceptions.InvalidSyntax(
+                    "'ingest-percentage' must be in the range (0.0, 100.0] but was {:.1f}".format(self.ingest_percentage))
+        except ValueError:
+            raise exceptions.InvalidSyntax("'ingest-percentage' must be numeric")
+
     def used_corpora(self, t, params):
         corpora = []
         track_corpora_names = [corpus.name for corpus in t.corpora]
@@ -504,7 +513,7 @@ class BulkIndexParamSource(ParamSource):
 
     def partition(self, partition_index, total_partitions):
         return PartitionBulkIndexParamSource(self.corpora, partition_index, total_partitions, self.batch_size, self.bulk_size,
-                                             self.id_conflicts, self.pipeline, self._params)
+                                             self.ingest_percentage, self.id_conflicts, self.pipeline, self._params)
 
     def params(self):
         raise exceptions.RallyError("Do not use a BulkIndexParamSource without partitioning")
@@ -514,7 +523,7 @@ class BulkIndexParamSource(ParamSource):
 
 
 class PartitionBulkIndexParamSource:
-    def __init__(self, corpora, partition_index, total_partitions, batch_size, bulk_size, id_conflicts=None,
+    def __init__(self, corpora, partition_index, total_partitions, batch_size, bulk_size, ingest_percentage, id_conflicts=None,
                  pipeline=None, original_params=None):
         """
 
@@ -523,14 +532,17 @@ class PartitionBulkIndexParamSource:
         :param total_partitions: The total number of partitions (i.e. clients) for bulk index operations.
         :param batch_size: The number of documents to read in one go.
         :param bulk_size: The size of bulk index operations (number of documents per bulk).
+        :param ingest_percentage: A number between (0.0, 100.0] that defines how much of the whole corpus should be ingested.
         :param id_conflicts: The type of id conflicts.
         :param pipeline: The name of the ingest pipeline to run.
+        :param original_params: The original dict passed to the parent parameter source.
         """
         self.corpora = corpora
         self.partition_index = partition_index
         self.total_partitions = total_partitions
         self.batch_size = batch_size
         self.bulk_size = bulk_size
+        self.ingest_percentage = ingest_percentage
         self.id_conflicts = id_conflicts
         self.pipeline = pipeline
         self.internal_params = bulk_data_based(total_partitions, partition_index, corpora, batch_size,
@@ -543,7 +555,8 @@ class PartitionBulkIndexParamSource:
         return next(self.internal_params)
 
     def size(self):
-        return number_of_bulks(self.corpora, self.partition_index, self.total_partitions, self.bulk_size)
+        all_bulks = number_of_bulks(self.corpora, self.partition_index, self.total_partitions, self.bulk_size)
+        return math.ceil((all_bulks * self.ingest_percentage) / 100)
 
 
 def number_of_bulks(corpora, partition_index, total_partitions, bulk_size):

Allow to ingest only a subset of the document corpus
For benchmarking smaller configurations (e.g. with very little memory), ingesting a full document corpus with tens of GB may be too much. For these cases we should allow users to provide a track parameter which allows them to define a percentage of the document corpus to ingest.

Notes: 

* Actually, this is already possible at the moment by forcing a specific number of iterations for the bulk operation. However, this is a bit hard to understand because the number of iterations is also tied to the bulk size.
* This is not meant to be accurate down to document level but only on a granularity of individual bulk requests (e.g. suppose we have a corpus with 1 million documents and a bulk size of 10.000 documents. If a user specifies 2.5% (= 25.000 documents) this will correspond to three bulk requests (i.e. 30.000 documents)).
diff --git a/tests/track/params_test.py b/tests/track/params_test.py
index 13794565..c6b9d8f2 100644
--- a/tests/track/params_test.py
+++ b/tests/track/params_test.py
@@ -475,11 +475,39 @@ class BulkIndexParamSourceTests(TestCase):
 
         self.assertEqual("Unknown 'conflicts' setting [crazy]", ctx.exception.args[0])
 
+    def test_create_with_ingest_percentage_too_low(self):
+        with self.assertRaises(exceptions.InvalidSyntax) as ctx:
+            params.BulkIndexParamSource(track=track.Track(name="unit-test"), params={
+                "bulk-size": 5000,
+                "ingest-percentage": 0.0
+            })
+
+        self.assertEqual("'ingest-percentage' must be in the range (0.0, 100.0] but was 0.0", ctx.exception.args[0])
+
+    def test_create_with_ingest_percentage_too_high(self):
+        with self.assertRaises(exceptions.InvalidSyntax) as ctx:
+            params.BulkIndexParamSource(track=track.Track(name="unit-test"), params={
+                "bulk-size": 5000,
+                "ingest-percentage": 100.1
+            })
+
+        self.assertEqual("'ingest-percentage' must be in the range (0.0, 100.0] but was 100.1", ctx.exception.args[0])
+
+    def test_create_with_ingest_percentage_not_numeric(self):
+        with self.assertRaises(exceptions.InvalidSyntax) as ctx:
+            params.BulkIndexParamSource(track=track.Track(name="unit-test"), params={
+                "bulk-size": 5000,
+                "ingest-percentage": "100 percent"
+            })
+
+        self.assertEqual("'ingest-percentage' must be numeric", ctx.exception.args[0])
+
     def test_create_valid_param_source(self):
         self.assertIsNotNone(params.BulkIndexParamSource(track.Track(name="unit-test"), params={
             "conflicts": "random",
             "bulk-size": 5000,
             "batch-size": 20000,
+            "ingest-percentage": 20.5,
             "pipeline": "test-pipeline"
         }))
 
@@ -565,6 +593,63 @@ class BulkIndexParamSourceTests(TestCase):
 
         self.assertEqual("The provided corpus ['does_not_exist'] does not match any of the corpora ['default'].", ctx.exception.args[0])
 
+    def test_ingests_all_documents_by_default(self):
+        corpora = [
+            track.DocumentCorpus(name="default", documents=[
+                track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK,
+                                number_of_documents=300000,
+                                target_index="test-idx",
+                                target_type="test-type"
+                                )
+            ]),
+            track.DocumentCorpus(name="special", documents=[
+                track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK,
+                                number_of_documents=700000,
+                                target_index="test-idx2",
+                                target_type="type"
+                                )
+            ]),
+        ]
+
+        source = params.BulkIndexParamSource(
+            track=track.Track(name="unit-test", corpora=corpora),
+            params={
+                "bulk-size": 10000
+            })
+
+        partition = source.partition(0, 1)
+        # # no ingest-percentage specified, should issue all one hundred bulk requests
+        self.assertEqual(100, partition.size())
+
+    def test_restricts_number_of_bulks_if_required(self):
+        corpora = [
+            track.DocumentCorpus(name="default", documents=[
+                track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK,
+                                number_of_documents=300000,
+                                target_index="test-idx",
+                                target_type="test-type"
+                                )
+            ]),
+            track.DocumentCorpus(name="special", documents=[
+                track.Documents(source_format=track.Documents.SOURCE_FORMAT_BULK,
+                                number_of_documents=700000,
+                                target_index="test-idx2",
+                                target_type="type"
+                                )
+            ]),
+        ]
+
+        source = params.BulkIndexParamSource(
+            track=track.Track(name="unit-test", corpora=corpora),
+            params={
+                "bulk-size": 10000,
+                "ingest-percentage": 2.5
+            })
+
+        partition = source.partition(0, 1)
+        # should issue three bulks of size 10.000
+        self.assertEqual(3, partition.size())
+
 
 class BulkDataGeneratorTests(TestCase):
     class TestBulkReader:

pip install -e .
parse_log_pytest
pytest
pytest pytest-benchmark
apt-get update apt-get install -y gcc python3-pip python3-dev
3.6
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
certifi==2021.5.30
elasticsearch==6.0.0
-e git+https://github.com/elastic/rally.git@7a0e577136d44360d61b3b91a0dc00ecc659f016#egg=esrally
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
Jinja2==2.9.5
jsonschema==2.5.1
MarkupSafe==2.0.1
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
psutil==5.4.0
py @ file:///opt/conda/conda-bld/py_1644396412707/work
py-cpuinfo==3.2.0
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
pytest-benchmark==3.4.1
tabulate==0.8.1
thespian==3.9.2
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3==1.22
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: rally
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - elasticsearch==6.0.0
      - jinja2==2.9.5
      - jsonschema==2.5.1
      - markupsafe==2.0.1
      - psutil==5.4.0
      - py-cpuinfo==3.2.0
      - pytest-benchmark==3.4.1
      - tabulate==0.8.1
      - thespian==3.9.2
      - urllib3==1.22
prefix: /opt/conda/envs/rally

tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_ingest_percentage_not_numeric tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_ingest_percentage_too_high tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_ingest_percentage_too_low tests/track/params_test.py::BulkIndexParamSourceTests::test_restricts_number_of_bulks_if_required
tests/track/params_test.py::SliceTests::test_slice_with_slice_larger_than_source tests/track/params_test.py::SliceTests::test_slice_with_source_larger_than_slice tests/track/params_test.py::ConflictingIdsBuilderTests::test_no_id_conflicts tests/track/params_test.py::ConflictingIdsBuilderTests::test_random_conflicts tests/track/params_test.py::ConflictingIdsBuilderTests::test_sequential_conflicts tests/track/params_test.py::ActionMetaDataTests::test_generate_action_meta_data_with_id_conflicts tests/track/params_test.py::ActionMetaDataTests::test_generate_action_meta_data_without_id_conflicts tests/track/params_test.py::ActionMetaDataTests::test_source_file_action_meta_data tests/track/params_test.py::IndexDataReaderTests::test_read_bulk_larger_than_number_of_docs tests/track/params_test.py::IndexDataReaderTests::test_read_bulk_smaller_than_number_of_docs tests/track/params_test.py::IndexDataReaderTests::test_read_bulk_smaller_than_number_of_docs_and_multiple_clients tests/track/params_test.py::IndexDataReaderTests::test_read_bulk_with_offset tests/track/params_test.py::IndexDataReaderTests::test_read_bulks_and_assume_metadata_line_in_source_file tests/track/params_test.py::InvocationGeneratorTests::test_build_conflicting_ids tests/track/params_test.py::InvocationGeneratorTests::test_calculate_bounds tests/track/params_test.py::InvocationGeneratorTests::test_calculate_non_multiple_bounds tests/track/params_test.py::InvocationGeneratorTests::test_calculate_number_of_bulks tests/track/params_test.py::InvocationGeneratorTests::test_iterator_chaining_respects_context_manager tests/track/params_test.py::BulkIndexParamSourceTests::test_create_valid_param_source tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_fraction_larger_batch_size tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_fraction_smaller_batch_size tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_metadata_in_source_file_but_conflicts tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_negative_bulk_size tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_non_numeric_bulk_size tests/track/params_test.py::BulkIndexParamSourceTests::test_create_with_unknown_id_conflicts tests/track/params_test.py::BulkIndexParamSourceTests::test_create_without_params tests/track/params_test.py::BulkIndexParamSourceTests::test_filters_corpora tests/track/params_test.py::BulkIndexParamSourceTests::test_ingests_all_documents_by_default tests/track/params_test.py::BulkIndexParamSourceTests::test_passes_all_corpora_by_default tests/track/params_test.py::BulkIndexParamSourceTests::test_raises_exception_if_no_corpus_matches tests/track/params_test.py::BulkDataGeneratorTests::test_generate_bulks_from_multiple_corpora tests/track/params_test.py::BulkDataGeneratorTests::test_generate_two_bulks tests/track/params_test.py::BulkDataGeneratorTests::test_internal_params_take_precedence tests/track/params_test.py::ParamsRegistrationTests::test_can_register_class_as_param_source tests/track/params_test.py::ParamsRegistrationTests::test_can_register_function_as_param_source tests/track/params_test.py::ParamsRegistrationTests::test_can_register_legacy_class_as_param_source tests/track/params_test.py::ParamsRegistrationTests::test_can_register_legacy_function_as_param_source tests/track/params_test.py::CreateIndexParamSourceTests::test_create_index_from_track_with_settings tests/track/params_test.py::CreateIndexParamSourceTests::test_create_index_from_track_without_settings tests/track/params_test.py::CreateIndexParamSourceTests::test_create_index_inline_with_body tests/track/params_test.py::CreateIndexParamSourceTests::test_create_index_inline_without_body tests/track/params_test.py::CreateIndexParamSourceTests::test_filter_index tests/track/params_test.py::DeleteIndexParamSourceTests::test_delete_index_by_name tests/track/params_test.py::DeleteIndexParamSourceTests::test_delete_index_from_track tests/track/params_test.py::DeleteIndexParamSourceTests::test_delete_no_index tests/track/params_test.py::DeleteIndexParamSourceTests::test_filter_index_from_track tests/track/params_test.py::CreateIndexTemplateParamSourceTests::test_create_index_template_from_track tests/track/params_test.py::CreateIndexTemplateParamSourceTests::test_create_index_template_inline tests/track/params_test.py::DeleteIndexTemplateParamSourceTests::test_delete_index_template_by_name tests/track/params_test.py::DeleteIndexTemplateParamSourceTests::test_delete_index_template_by_name_and_matching_indices tests/track/params_test.py::DeleteIndexTemplateParamSourceTests::test_delete_index_template_by_name_and_matching_indices_missing_index_pattern tests/track/params_test.py::DeleteIndexTemplateParamSourceTests::test_delete_index_template_from_track tests/track/params_test.py::SearchParamSourceTests::test_passes_request_parameters tests/track/params_test.py::SearchParamSourceTests::test_replaces_body_params
docs/track.rst esrally/track/params.py
docs/track.rst esrally/track/params.py