diff --git a/README.md b/README.md
index 799f5e8d..1a5c3764 100644
--- a/README.md
+++ b/README.md
@@ -139,6 +139,7 @@ composer-dev create \
   --port WEB_SERVER_PORT \
   --dags-path LOCAL_DAGS_PATH \
   --plugins-path LOCAL_PLUGINS_PATH \
+  --data-path LOCAL_DATA_PATH \
   --database DATABASE_ENGINE \
   LOCAL_ENVIRONMENT_NAME
 ```
@@ -152,6 +153,9 @@ Replace:
     located.
 - `LOCAL_PLUGINS_PATH` with the path to a local directory where the plugins
     files are located.
+- `LOCAL_DATA_PATH` with the path to a local directory where the data files are
+    located. If omitted, a `data` directory inside the environment directory is
+    used.
 - `DATABASE_ENGINE` with the database engine you wanted to use. You can use
     `sqlite` or `postgresql` (default).
 - `LOCAL_ENVIRONMENT_NAME` with the name of this local Airflow environment.
@@ -200,7 +204,8 @@ composer-dev create LOCAL_ENVIRONMENT_NAME \
     --project PROJECT_ID \
     --port WEB_SERVER_PORT \
     --dags-path LOCAL_DAGS_PATH \
-    --plugins-path LOCAL_PLUGINS_PATH
+    --plugins-path LOCAL_PLUGINS_PATH \
+    --data-path LOCAL_DATA_PATH
 ```
 
 Replace:
@@ -215,6 +220,9 @@ Replace:
     located.
 - `LOCAL_PLUGINS_PATH` with a path to a local directory where the plugins are
     located.
+- `LOCAL_DATA_PATH` with a path to a local directory where the data files are
+    located. If omitted, a `data` directory inside the environment directory is
+    used.
 
 Example:
 
@@ -302,17 +310,20 @@ composer-dev stop LOCAL_ENVIRONMENT_NAME
 
 ## Add and update DAGs
 
-DAGs and plugins are stored in the directories that you specified in the
-`--dags-path` and `--plugins-path` parameters respectively when you created
-your local Airflow environment. By default, these directories are
-`./composer/<local_environment_name>/dags` and
-`./composer/<local_environment_name>/plugins`.
+DAGs, plugins, and data are stored in the directories that you specified in the
+`--dags-path`, `--plugins-path`, and `--data-path` parameters respectively when
+you created your local Airflow environment. By default, these directories are
+`./composer/<local_environment_name>/dags`,
+`./composer/<local_environment_name>/plugins`, and
+`./composer/<local_environment_name>/data`. The data directory is mounted into
+the container at `/home/airflow/gcs/data`.
 
 You can get the directories used by your environment with the
 [`describe` command](#get-a-list-and-status-of-local-airflow-environments).
 
-To add and update DAGs and plugins, change files in these directories. You do
-not need to restart your local Airflow environment for changes to take effect.
+To add and update DAGs, plugins, and data, change files in these directories.
+You do not need to restart your local Airflow environment for changes to take
+effect.
 
 ## View local Airflow environment logs
 
diff --git a/composer_local_dev/cli.py b/composer_local_dev/cli.py
index bc8cbda9..8b58c53f 100644
--- a/composer_local_dev/cli.py
+++ b/composer_local_dev/cli.py
@@ -48,7 +48,12 @@
         },
         {
             "name": "Environment options",
-            "options": ["--web-server-port", "--dags-path", "--plugins-path"],
+            "options": [
+                "--web-server-port",
+                "--dags-path",
+                "--plugins-path",
+                "--data-path",
+            ],
         },
         {
             "name": "Container Memory and CPUs limit",
@@ -260,6 +265,13 @@ def _complete_environment(ctx, param, incomplete):
     metavar="PATH",
     type=click.Path(file_okay=False),
 )
+@click.option(
+    "--data-path",
+    help="Path to data folder. If it does not exist, it will be created.",
+    show_default="'data' directory in the environment directory",
+    metavar="PATH",
+    type=click.Path(file_okay=False),
+)
 @click.option(
     "--database-engine",
     "--database",
@@ -285,6 +297,7 @@ def create(
     database_engine: str,
     dags_path: Optional[pathlib.Path] = None,
     plugins_path: Optional[pathlib.Path] = None,
+    data_path: Optional[pathlib.Path] = None,
     container_memory_limit: Optional[str] = None,
     container_cpu_limit: Optional[str] = None,
 ):
@@ -341,6 +354,7 @@ def create(
             web_server_port=web_server_port,
             dags_path=dags_path,
             plugins_path=plugins_path,
+            data_path=data_path,
             database_engine=database_engine,
             memory_limit=container_memory_limit,
             cpu_count=container_cpu_limit,
@@ -354,6 +368,7 @@ def create(
             port=web_server_port,
             dags_path=dags_path,
             plugins_path=plugins_path,
+            data_path=data_path,
             database_engine=database_engine,
             memory_limit=container_memory_limit,
             cpu_count=container_cpu_limit,
diff --git a/composer_local_dev/constants.py b/composer_local_dev/constants.py
index e470b078..cd68ee6c 100644
--- a/composer_local_dev/constants.py
+++ b/composer_local_dev/constants.py
@@ -94,7 +94,8 @@ def choices(cls):
 
 1. You can put your DAGs in {dags_path}
 2. You can put your plugins in {plugins_path}
-3. Access Airflow at http://localhost:{port}
+3. You can put your data in {data_path}
+4. Access Airflow at http://localhost:{port}
 """
 # TODO: Fill source environment info
 DESCRIBE_ENV_MESSAGE = """
@@ -103,6 +104,7 @@ def choices(cls):
 Image version: {image_version}
 Dags directory: {dags_path}.
 Plugins directory: {plugins_path}.
+Data directory: {data_path}.
 The environment is using credentials from gcloud located at {gcloud_path}.
 """
 KUBECONFIG_PATH_MESSAGE = """
@@ -142,6 +144,15 @@ def choices(cls):
 PLUGINS_PATH_NOT_EXISTS_ERROR = (
     "Plugins path does not exist or is not a directory: {plugins_path}"
 )
+CREATING_DATA_PATH_WARN = (
+    "Data path '{data_path}' does not exist. It will be created."
+)
+DATA_PATH_NOT_PROVIDED_WARN = (
+    "No data directory provided, using default data directory."
+)
+DATA_PATH_NOT_EXISTS_ERROR = (
+    "Data path does not exist or is not a directory: {data_path}"
+)
 FAILED_TO_GET_DOCKER_PORT_WARN = (
     "Failed to retrieve used port from the Docker daemon, "
     "using port from the environment configuration."
diff --git a/composer_local_dev/environment.py b/composer_local_dev/environment.py
index 6326d29a..6d05ea86 100644
--- a/composer_local_dev/environment.py
+++ b/composer_local_dev/environment.py
@@ -52,6 +52,7 @@ def get_image_mounts(
     kube_config_path: Optional[str],
     requirements: pathlib.Path,
     database_mounts: Dict[pathlib.Path, str],
+    data_path: Optional[str] = None,
 ) -> List[docker.types.Mount]:
     """
     Return list of docker volumes to be mounted inside container.
@@ -63,11 +64,14 @@ def get_image_mounts(
      - environment airflow sqlite db file location
      - database_mounts which contains the path for database mounts
     """
+    # Backwards compatibility: when no explicit data_path is given, the data
+    # directory lives inside the environment directory (the original behaviour).
+    data_path = data_path if data_path is not None else env_path / "data"
     mount_paths = {
         requirements: "composer_requirements.txt",
         dags_path: "gcs/dags/",
         plugins_path: "gcs/plugins/",
-        env_path / "data": "gcs/data/",
+        data_path: "gcs/data/",
         gcloud_config_path: ".config/gcloud",
         **database_mounts,
     }
@@ -373,6 +377,11 @@ def __init__(self, env_dir_path: pathlib.Path, port: Optional[int]):
             self.plugins_path = self.get_str_param("plugins_path")
         else:
             self.plugins_path = files.resolve_plugins_path(None, env_dir_path)
+        # Backwards compatibility: don't fail on missing data_path
+        if "data_path" in self.config:
+            self.data_path = self.get_str_param("data_path")
+        else:
+            self.data_path = files.resolve_data_path(None, env_dir_path)
         self.dag_dir_list_interval = self.parse_int_param(
             "dag_dir_list_interval", allowed_range=(0,)
         )
@@ -457,6 +466,7 @@ def __init__(
         location: str,
         dags_path: Optional[str],
         plugins_path: Optional[str] = None,
+        data_path: Optional[str] = None,
         dag_dir_list_interval: int = 10,
         database_engine: str = constants.DatabaseEngine.postgresql,
         memory_limit: Optional[str] = None,
@@ -488,6 +498,7 @@ def __init__(
         self.plugins_path = files.resolve_plugins_path(
             plugins_path, env_dir_path
         )
+        self.data_path = files.resolve_data_path(data_path, env_dir_path)
         self.dag_dir_list_interval = dag_dir_list_interval
         self.database_engine = database_engine
         self.is_database_sqlite3 = (
@@ -563,6 +574,7 @@ def load_from_config(cls, env_dir_path: pathlib.Path, port: Optional[int]):
             location=config.location,
             dags_path=config.dags_path,
             plugins_path=config.plugins_path,
+            data_path=config.data_path,
             dag_dir_list_interval=config.dag_dir_list_interval,
             port=config.port,
             database_engine=config.database_engine,
@@ -582,6 +594,7 @@ def from_source_environment(
         dags_path: Optional[str],
         plugins_path: Optional[str],
         database_engine: str,
+        data_path: Optional[str] = None,
         memory_limit: Optional[str] = None,
         cpu_count: Optional[int] = None,
     ):
@@ -606,6 +619,7 @@ def from_source_environment(
             location=location,
             dags_path=dags_path,
             plugins_path=plugins_path,
+            data_path=data_path,
             dag_dir_list_interval=10,
             port=web_server_port,
             pypi_packages=pypi_packages,
@@ -720,6 +734,7 @@ def write_environment_config_to_config_file(self):
             "composer_project_id": self.project_id,
             "dags_path": self.dags_path,
             "plugins_path": self.plugins_path,
+            "data_path": self.data_path,
             "dag_dir_list_interval": int(self.dag_dir_list_interval),
             "port": int(self.port),
             "database_engine": self.database_engine,
@@ -802,6 +817,7 @@ def create_docker_container(self):
             utils.resolve_kube_config_path(),
             self.requirements_file,
             db_mounts,
+            self.data_path,
         )
         db_vars = db_extras["env_vars"]
         default_vars = self.get_default_environment_variables(db_vars)
@@ -886,6 +902,7 @@ def create_db_docker_container(self):
             utils.resolve_kube_config_path(),
             self.requirements_file,
             db_mounts,
+            self.data_path,
         )
         db_vars = db_extras["env_vars"]
         db_ports = db_extras["ports"]
@@ -967,7 +984,10 @@ def create(self):
         assert_image_exists(self.image_version)
         self.assert_valid_environment_options()
         files.create_environment_directories(
-            self.env_dir_path, self.dags_path, self.plugins_path
+            self.env_dir_path,
+            self.dags_path,
+            self.plugins_path,
+            self.data_path,
         )
         self.create_database_files(skip_if_exist=False)
         self.write_environment_config_to_config_file()
@@ -1104,6 +1124,7 @@ def start(self, assert_not_running=True):
         self.assert_requirements_exist()
         files.assert_dag_path_exists(self.dags_path)
         files.assert_plugins_path_exists(self.plugins_path)
+        files.assert_data_path_exists(self.data_path)
 
         self.create_database_files()
         db_path = (
@@ -1153,6 +1174,7 @@ def print_start_message(self):
                 env_name=self.name,
                 dags_path=self.dags_path,
                 plugins_path=self.plugins_path,
+                data_path=self.data_path,
                 port=self.port,
             )
         )
@@ -1278,6 +1300,7 @@ def prepare_env_description(self, env_status: str) -> str:
                 image_version=self.image_version,
                 dags_path=self.dags_path,
                 plugins_path=self.plugins_path,
+                data_path=self.data_path,
                 gcloud_path=utils.resolve_gcloud_config_path(),
             )
             + (
diff --git a/composer_local_dev/errors.py b/composer_local_dev/errors.py
index 0e899cb5..24f31b7d 100644
--- a/composer_local_dev/errors.py
+++ b/composer_local_dev/errors.py
@@ -191,6 +191,15 @@ def __init__(self, plugins_path):
         )
 
 
+class DataPathNotExistError(ComposerCliError):
+    """Data path does not exist or is not a directory."""
+
+    def __init__(self, data_path):
+        super().__init__(
+            constants.DATA_PATH_NOT_EXISTS_ERROR.format(data_path=data_path)
+        )
+
+
 def catch_exceptions(func=None):
     """
     Catch exceptions and print user friendly message for common issues.
diff --git a/composer_local_dev/files.py b/composer_local_dev/files.py
index 5f7bf8bc..673c88dc 100644
--- a/composer_local_dev/files.py
+++ b/composer_local_dev/files.py
@@ -126,8 +126,28 @@ def resolve_plugins_path(
     return str(plugins_path.resolve())
 
 
+def resolve_data_path(data_path: Optional[str], env_dir: pathlib.Path) -> str:
+    """
+    Provides and validates path to the data directory.
+    If ``data_path`` is None, the path is constructed from ``env_dir`` path
+    and ``data`` directory.
+    If ``data_path`` is not None, but it does not exist, a warning is raised.
+
+    Returns absolute ``data_path`` path.
+    """
+    if data_path is None:
+        console.get_console().print(constants.DATA_PATH_NOT_PROVIDED_WARN)
+        data_path = env_dir / "data"
+    else:
+        data_path = pathlib.Path(data_path)
+    return str(data_path.resolve())
+
+
 def create_environment_directories(
-    env_dir: pathlib.Path, dags_path: str, plugins_path: str
+    env_dir: pathlib.Path,
+    dags_path: str,
+    plugins_path: str,
+    data_path: Optional[str] = None,
 ):
     """
     Create environment directories (overwriting existing ones).
@@ -135,14 +155,17 @@ def create_environment_directories(
     composer local environment and files used by environment such as
     requirements.txt file, dags, data and plugins directories.
     """
-    data_dir = "data"
-    LOG.info(
-        "Creating environment directory %s in " "%s environment directory.",
-        data_dir,
-        env_dir,
+    # Backwards compatibility: when no explicit data_path is given, the data
+    # directory lives inside the environment directory (the original behaviour).
+    data_path = (
+        pathlib.Path(data_path) if data_path is not None else env_dir / "data"
     )
     env_dir.mkdir(exist_ok=True, parents=True)
-    (env_dir / data_dir).mkdir(exist_ok=True)
+    if not data_path.is_dir():
+        console.get_console().print(
+            constants.CREATING_DATA_PATH_WARN.format(data_path=data_path)
+        )
+        data_path.mkdir(parents=True)
 
     dags_path = pathlib.Path(dags_path)
     if not dags_path.is_dir():
@@ -259,3 +282,10 @@ def assert_plugins_path_exists(path: str) -> None:
     if pathlib.Path(path).is_dir():
         return
     raise errors.PluginsPathNotExistError(path)
+
+
+def assert_data_path_exists(path: str) -> None:
+    """Raise an error if data path does not point to existing directory."""
+    if pathlib.Path(path).is_dir():
+        return
+    raise errors.DataPathNotExistError(path)
diff --git a/tests/e2e/incompatible_requirements.txt b/tests/e2e/incompatible_requirements.txt
index b1d81767..f6cfce96 100644
--- a/tests/e2e/incompatible_requirements.txt
+++ b/tests/e2e/incompatible_requirements.txt
@@ -1,4 +1,4 @@
 protobuf==5.29.6
 google-cloud-dlp==3.9.2
-apache-airflow-providers-google==8.6.0
+apache-airflow-providers-google==8.10.0
 proto-plus==1.20.5
\ No newline at end of file
diff --git a/tests/unit/test_environment.py b/tests/unit/test_environment.py
index ea771d8b..8d497442 100644
--- a/tests/unit/test_environment.py
+++ b/tests/unit/test_environment.py
@@ -34,7 +34,10 @@
 @mock.patch("composer_local_dev.environment.docker.from_env")
 @mock.patch("composer_local_dev.environment.files.resolve_dags_path")
 @mock.patch("composer_local_dev.environment.files.resolve_plugins_path")
-def default_env(mocked_docker, mocked_dags, mocked_plugins, tmp_path):
+@mock.patch("composer_local_dev.environment.files.resolve_data_path")
+def default_env(
+    mocked_data, mocked_docker, mocked_dags, mocked_plugins, tmp_path
+):
     env_dir_path = tmp_path / ".compose" / "my_env"
     env = environment.Environment(
         env_dir_path=env_dir_path,
@@ -52,8 +55,9 @@ def default_env(mocked_docker, mocked_dags, mocked_plugins, tmp_path):
 @mock.patch("composer_local_dev.environment.docker.from_env")
 @mock.patch("composer_local_dev.environment.files.resolve_dags_path")
 @mock.patch("composer_local_dev.environment.files.resolve_plugins_path")
+@mock.patch("composer_local_dev.environment.files.resolve_data_path")
 def default_env_postgresql(
-    mocked_docker, mocked_dags, mocked_plugins, tmp_path
+    mocked_data, mocked_docker, mocked_dags, mocked_plugins, tmp_path
 ):
     env_dir_path = tmp_path / ".compose" / "my_env"
     env = environment.Environment(
@@ -830,6 +834,7 @@ def test_create_db_docker_container(
     @pytest.mark.parametrize(
         "container_exists, create_container", [(False, True), (True, False)]
     )
+    @mock.patch("composer_local_dev.files.assert_data_path_exists")
     @mock.patch("composer_local_dev.files.assert_dag_path_exists")
     @mock.patch("composer_local_dev.files.assert_plugins_path_exists")
     @mock.patch("composer_local_dev.environment.assert_image_exists")
@@ -844,6 +849,7 @@ def test_start_container(
         mocked_create,
         mocked_fix,
         mocked_line_fix,
+        mocked_data_assert,
         container_exists,
         create_container,
         default_env,
@@ -869,6 +875,7 @@ def test_start_container(
         )
         default_env.wait_for_start.assert_called_once()
 
+    @mock.patch("composer_local_dev.files.assert_data_path_exists")
     @mock.patch("composer_local_dev.files.assert_dag_path_exists")
     @mock.patch("composer_local_dev.files.assert_plugins_path_exists")
     @mock.patch("composer_local_dev.environment.assert_image_exists")
@@ -883,6 +890,7 @@ def test_start_already_running(
         mocked_create,
         mocked_fix,
         mocked_line_endings,
+        mocked_data_assert,
         default_env,
     ):
         default_env.assert_requirements_exist = mock.Mock()
@@ -897,6 +905,7 @@ def test_start_already_running(
         ):
             default_env.start()
 
+    @mock.patch("composer_local_dev.files.assert_data_path_exists")
     @mock.patch("composer_local_dev.files.assert_dag_path_exists")
     @mock.patch("composer_local_dev.files.assert_plugins_path_exists")
     @mock.patch("composer_local_dev.environment.assert_image_exists")
@@ -911,6 +920,7 @@ def test_restart_already_running(
         mocked_create,
         mocked_fix,
         mocked_line_endings,
+        mocked_data_assert,
         default_env,
     ):
         default_env.assert_requirements_exist = mock.Mock()
@@ -947,6 +957,7 @@ def test_prepare_env_description_running(
             image_version=default_env.image_version,
             dags_path=default_env.dags_path,
             plugins_path=default_env.plugins_path,
+            data_path=default_env.data_path,
             gcloud_path="path",
         )
         kub_desc = constants.KUBECONFIG_PATH_MESSAGE.format(
@@ -976,6 +987,7 @@ def test_prepare_env_description_not_running(
             image_version=default_env.image_version,
             dags_path=default_env.dags_path,
             plugins_path=default_env.plugins_path,
+            data_path=default_env.data_path,
             gcloud_path="path",
             kube_config_path="path/kube",
         )
@@ -1068,6 +1080,7 @@ def test_create_docker_container_pull_not_existing_image(
         default_env.create_docker_container()
         default_env.pull_image.assert_called_once()
 
+    @mock.patch("composer_local_dev.files.assert_data_path_exists")
     @mock.patch("composer_local_dev.files.assert_dag_path_exists")
     @mock.patch("composer_local_dev.files.assert_plugins_path_exists")
     @mock.patch("composer_local_dev.environment.assert_image_exists")
@@ -1082,6 +1095,7 @@ def test_start_used_port(
         mocked_create,
         mocked_fix,
         mocked_line_fix,
+        mocked_data_assert,
         default_env,
     ):
         default_env.port = 8083
@@ -1409,6 +1423,42 @@ def test_get_image_mounts(mocked_mount):
     mocked_mount.assert_has_calls(expected_mounts)
 
 
+@mock.patch("composer_local_dev.environment.docker.types.Mount", autospec=True)
+def test_get_image_mounts_with_custom_data_path(mocked_mount):
+    path = pathlib.Path("path/dir")
+    dags_path = "path/to/dags"
+    plugins_path = "path/to/plugins"
+    data_path = "custom/path/to/data"
+    gcloud_path = "config/path"
+    kubeconfig_path = "/kube"
+    requirements = path / "requirements.txt"
+    environment.get_image_mounts(
+        path,
+        dags_path,
+        plugins_path,
+        gcloud_path,
+        kubeconfig_path,
+        requirements,
+        {},
+        data_path,
+    )
+    # Explicit data_path is mounted to the gcs/data target ...
+    mocked_mount.assert_any_call(
+        source=data_path,
+        target="/home/airflow/gcs/data/",
+        type="bind",
+    )
+    # ... and the default <env_dir>/data fallback is NOT used.
+    assert (
+        mock.call(
+            source=str(path / "data"),
+            target="/home/airflow/gcs/data/",
+            type="bind",
+        )
+        not in mocked_mount.mock_calls
+    )
+
+
 @mock.patch("composer_local_dev.environment.service_v1", autospec=True)
 def test_get_software_config_from_environment_api_error(mocked_service):
     error_msg = "Foo error"
@@ -1566,3 +1616,42 @@ def test_param_invalid_int_range(
         with pytest.raises(errors.FailedToParseConfigParamIntRangeError) as err:
             environment.EnvironmentConfig(tmp_path, None)
             assert str(err) == exp_error
+
+    @mock.patch(
+        "composer_local_dev.environment.EnvironmentConfig.load_configuration_from_file"
+    )
+    def test_data_path_from_config(self, mocked_load_conf, tmp_path):
+        config = {
+            "composer_image_version": "composer-2.0.25-airflow-2.2.5",
+            "composer_location": "us-central1",
+            "composer_project_id": "project",
+            "dags_path": "/dags/",
+            "plugins_path": "/plugins/",
+            "data_path": "/custom/data/",
+            "dag_dir_list_interval": 10,
+            "port": 8080,
+            "database_engine": "postgresql",
+        }
+        mocked_load_conf.return_value = config
+        env_config = environment.EnvironmentConfig(tmp_path, None)
+        assert env_config.data_path == "/custom/data/"
+
+    @mock.patch(
+        "composer_local_dev.environment.EnvironmentConfig.load_configuration_from_file"
+    )
+    def test_data_path_defaults_when_missing(self, mocked_load_conf, tmp_path):
+        # Backwards compatibility: configs without a data_path fall back to
+        # the 'data' directory inside the environment directory.
+        config = {
+            "composer_image_version": "composer-2.0.25-airflow-2.2.5",
+            "composer_location": "us-central1",
+            "composer_project_id": "project",
+            "dags_path": "/dags/",
+            "plugins_path": "/plugins/",
+            "dag_dir_list_interval": 10,
+            "port": 8080,
+            "database_engine": "postgresql",
+        }
+        mocked_load_conf.return_value = config
+        env_config = environment.EnvironmentConfig(tmp_path, None)
+        assert env_config.data_path == str((tmp_path / "data").resolve())
diff --git a/tests/unit/test_files.py b/tests/unit/test_files.py
index 2ba50797..76018c97 100644
--- a/tests/unit/test_files.py
+++ b/tests/unit/test_files.py
@@ -174,6 +174,10 @@ def test_not_existing_dag_path(self, temporary_env_dir, capsys):
         dags_path = env_dir / "dags"
         plugins_path = env_dir / "plugins"
         plugins_path.mkdir()
+        # Pre-create the default data dir so this test stays focused on the
+        # dags path warning (an absent data dir warns on its own, see
+        # test_not_existing_data_path).
+        (env_dir / "data").mkdir()
         files.create_environment_directories(
             env_dir=env_dir, dags_path=dags_path, plugins_path=plugins_path
         )
@@ -211,6 +215,10 @@ def test_not_existing_plugins_path(self, temporary_env_dir, capsys):
         dags_path = env_dir / "dags"
         dags_path.mkdir()
         plugins_path = env_dir / "plugins"
+        # Pre-create the default data dir so this test stays focused on the
+        # plugins path warning (an absent data dir warns on its own, see
+        # test_not_existing_data_path).
+        (env_dir / "data").mkdir()
         files.create_environment_directories(
             env_dir=env_dir, dags_path=dags_path, plugins_path=plugins_path
         )
@@ -229,6 +237,48 @@ def test_not_existing_plugins_path(self, temporary_env_dir, capsys):
             assert path.is_dir()
         assert expected_output == output
 
+    def test_not_existing_data_path(self, temporary_env_dir, capsys):
+        env_dir = temporary_env_dir
+        dags_path = env_dir / "dags"
+        dags_path.mkdir()
+        plugins_path = env_dir / "plugins"
+        plugins_path.mkdir()
+        files.create_environment_directories(
+            env_dir=env_dir, dags_path=dags_path, plugins_path=plugins_path
+        )
+        captured = capsys.readouterr()
+        output = clean_cli_output(captured.out)
+        expected_output = clean_cli_output(
+            constants.CREATING_DATA_PATH_WARN.format(data_path=env_dir / "data")
+        )
+        assert (env_dir / "data").is_dir()
+        assert expected_output == output
+
+    def test_custom_data_path(self, temporary_env_dir, capsys):
+        env_dir = temporary_env_dir
+        dags_path = env_dir / "dags"
+        dags_path.mkdir()
+        plugins_path = env_dir / "plugins"
+        plugins_path.mkdir()
+        data_path = env_dir / "custom_data"
+        files.create_environment_directories(
+            env_dir=env_dir,
+            dags_path=dags_path,
+            plugins_path=plugins_path,
+            data_path=data_path,
+        )
+        captured = capsys.readouterr()
+        output = clean_cli_output(captured.out)
+        expected_output = clean_cli_output(
+            constants.CREATING_DATA_PATH_WARN.format(data_path=data_path)
+        )
+        assert env_dir.is_dir()
+        assert data_path.is_dir()
+        # The default <env_dir>/data is not created when data_path is given.
+        assert not (env_dir / "data").exists()
+        # The user is told the custom data path will be created.
+        assert expected_output == output
+
 
 class TestResolveDagsPath:
     def test_optional_dags_path(self, tmpdir, capsys):
@@ -268,6 +318,24 @@ def test_provided_plugins_path(self, tmpdir):
         assert expected_plugins_str_path == plugins_path
 
 
+class TestResolveDataPath:
+    def test_optional_data_path(self, tmpdir, capsys):
+        env_dir = pathlib.Path(tmpdir)
+        expected_data_path = str(env_dir / "data")
+        data_path = files.resolve_data_path(None, env_dir)
+        captured = capsys.readouterr()
+        output = "".join(captured.out.split("\n"))
+        assert expected_data_path == data_path
+        assert constants.DATA_PATH_NOT_PROVIDED_WARN in output
+
+    def test_provided_data_path(self, tmpdir):
+        expected_data_path = pathlib.Path(tmpdir) / "data"
+        expected_data_path.mkdir(exist_ok=True)
+        expected_data_str_path = str(expected_data_path)
+        data_path = files.resolve_data_path(expected_data_str_path, tmpdir)
+        assert expected_data_str_path == data_path
+
+
 class TestAssertDagsPathExists:
     def test_existing_path(self, tmp_path):
         files.assert_dag_path_exists(str(tmp_path))
@@ -314,6 +382,29 @@ def test_path_is_file(self, tmp_path):
             assert str(err) == error_msg
 
 
+class TestAssertDataPathExists:
+    def test_existing_path(self, tmp_path):
+        files.assert_data_path_exists(str(tmp_path))
+
+    def test_missing_path(self):
+        path = "i/dont/exist"
+        error_msg = f"Data path does not exist or is not a directory: {path}"
+        with pytest.raises(errors.DataPathNotExistError) as err:
+            files.assert_data_path_exists(path)
+            assert str(err) == error_msg
+
+    def test_path_is_file(self, tmp_path):
+        file_path = tmp_path / "file.ext"
+        with open(file_path, "w") as fp:
+            pass
+        error_msg = (
+            f"Data path does not exist or is not a directory: {file_path}"
+        )
+        with pytest.raises(errors.DataPathNotExistError) as err:
+            files.assert_data_path_exists(str(file_path))
+            assert str(err) == error_msg
+
+
 def test_dos2unix_file(tmp_path):
     windows_string = b"file\r\nwith carriage returns\r\n"
     linux_string = b"file\nwith carriage returns\n"