diff --git a/BUILD b/BUILD
index 3dd5093ab..f5f8dab8b 100644
--- a/BUILD
+++ b/BUILD
@@ -28,6 +28,7 @@ test_suite(
         "//mlvm:test_mlvm",
         "//oozie:test_oozie",
         "//presto:test_presto",
+        "//python:test_python",
         "//ranger:test_ranger",
         "//rapids:test_rapids",
         "//rstudio:test_rstudio",
diff --git a/python/BUILD b/python/BUILD
new file mode 100644
index 000000000..aff430881
--- /dev/null
+++ b/python/BUILD
@@ -0,0 +1,38 @@
+package(default_visibility = ["//visibility:public"])
+
+
+py_test(
+    name = "test_pip_cached",
+    size = "enormous",
+    srcs = ["test_pip_cached.py"],
+    data = [
+        "pip-install-cached.sh",
+    ],
+    local = True,
+    deps = [
+        "//integration_tests:dataproc_test_case",
+        "@io_abseil_py//absl/testing:parameterized",
+    ],
+)
+
+py_test(
+    name = "test_conda_cached",
+    size = "enormous",
+    srcs = ["test_conda_cached.py"],
+    data = [
+        "conda-install-cached.sh",
+    ],
+    local = True,
+    deps = [
+        "//integration_tests:dataproc_test_case",
+        "@io_abseil_py//absl/testing:parameterized",
+    ],
+)
+
+test_suite(
+    name = "test_python",
+    tests = [
+        ":test_pip_cached",
+        ":test_conda_cached",
+    ],
+)
diff --git a/python/README.md b/python/README.md
index 682c2db1d..133aca636 100644
--- a/python/README.md
+++ b/python/README.md
@@ -94,3 +94,56 @@ gcloud dataproc clusters create ${CLUSTER_NAME} \
     --metadata 'CONDA_PACKAGES=recentrifuge=1.0.0' \
     --initialization-actions gs://goog-dataproc-initialization-actions-${REGION}/python/conda-install.sh
 ```
+
+## Install PyPI packages with Caching
+
+Install [PyPI](https://pypi.org) packages into user Python environment using `pip` command, with support for caching the resulting environment in GCS to speed up subsequent cluster creations.
+
+### Options
+
+-   `PIP_PACKAGES` - a space separated list of packages to install. Packages can contain version selectors.
+-   `CACHE_BUCKET` - (Optional) The GCS bucket to store/fetch caches. Defaults to the cluster's temporary bucket (`dataproc-temp-bucket`).
+-   `CACHE_KEY_OVERRIDE` - (Optional) Explicit name for the cache file. If omitted, a hash of the packages and environment is used.
+-   `CACHE_TIMEOUT` - (Optional) Cache validity period in seconds. Defaults to 10 years (effectively infinite).
+-   `TARGET_ENV_PATH` - (Optional) Explicit path to the target Python environment. If omitted, defaults to the active environment.
+
+### Examples
+
+Installing packages with default caching (uses temp bucket):
+
+```bash
+REGION=<region>
+CLUSTER_NAME=<cluster_name>
+gcloud dataproc clusters create ${CLUSTER_NAME} \
+    --region ${REGION} \
+    --metadata 'PIP_PACKAGES=pandas scipy' \
+    --initialization-actions gs://goog-dataproc-initialization-actions-${REGION}/python/pip-install-cached.sh
+```
+
+## Install Conda packages with Caching
+
+Install Conda packages into user Python environment using `conda` command, with support for caching the resulting environment in GCS to speed up subsequent cluster creations.
+
+### Options
+
+-   `CONDA_CHANNELS` - a space separated list of new channels to configure in addition to default channels.
+-   `CONDA_PACKAGES` - a space separated list of packages to install. Packages can contain version selectors.
+-   `CACHE_BUCKET` - (Optional) The GCS bucket to store/fetch caches. Defaults to the cluster's temporary bucket (`dataproc-temp-bucket`).
+-   `CACHE_KEY_OVERRIDE` - (Optional) Explicit name for the cache file. If omitted, a hash of the channels, packages, and environment is used.
+-   `CACHE_TIMEOUT` - (Optional) Cache validity period in seconds. Defaults to 10 years (effectively infinite).
+-   `TARGET_ENV_PATH` - (Optional) Explicit path to the target Conda environment. If omitted, defaults to the active environment.
+
+### Examples
+
+Installing packages with explicit cache bucket and override:
+
+```bash
+REGION=<region>
+CLUSTER_NAME=<cluster_name>
+gcloud dataproc clusters create ${CLUSTER_NAME} \
+    --region ${REGION} \
+    --metadata 'CONDA_PACKAGES=scipy' \
+    --metadata 'CACHE_BUCKET=my-custom-cache-bucket' \
+    --metadata 'CACHE_KEY_OVERRIDE=my-scipy-env' \
+    --initialization-actions gs://goog-dataproc-initialization-actions-${REGION}/python/conda-install-cached.sh
+```
diff --git a/python/conda-install-cached.sh b/python/conda-install-cached.sh
new file mode 100755
index 000000000..a16a4b80a
--- /dev/null
+++ b/python/conda-install-cached.sh
@@ -0,0 +1,329 @@
+#!/bin/bash
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS-IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This script installs Conda packages, using GCS to cache the resulting environment.
+
+set -exo pipefail
+
+readonly CHANNELS=$(/usr/share/google/get_metadata_value attributes/CONDA_CHANNELS || true)
+readonly PACKAGES=$(/usr/share/google/get_metadata_value attributes/CONDA_PACKAGES || true)
+CACHE_BUCKET=$(/usr/share/google/get_metadata_value attributes/CACHE_BUCKET || true)
+if [[ -z "${CACHE_BUCKET}" ]]; then
+  CACHE_BUCKET=$(/usr/share/google/get_metadata_value attributes/dataproc-temp-bucket || true)
+  if [[ -n "${CACHE_BUCKET}" ]]; then
+    echo "CACHE_BUCKET not specified, defaulting to dataproc-temp-bucket: ${CACHE_BUCKET}"
+  fi
+fi
+readonly CACHE_BUCKET
+readonly CACHE_KEY_OVERRIDE=$(/usr/share/google/get_metadata_value attributes/CACHE_KEY_OVERRIDE || true)
+CACHE_TIMEOUT=$(/usr/share/google/get_metadata_value attributes/CACHE_TIMEOUT || true)
+if [[ -z "${CACHE_TIMEOUT}" ]]; then
+  CACHE_TIMEOUT=315360000 # 10 years (effectively infinite)
+fi
+readonly CACHE_TIMEOUT
+readonly TARGET_ENV_PATH=$(/usr/share/google/get_metadata_value attributes/TARGET_ENV_PATH || true)
+readonly OS_NAME=$(grep '^ID=' /etc/os-release | cut -d= -f2 | xargs | tr '[:upper:]' '[:lower:]')
+
+GCS_CMD="gsutil"
+if gcloud --help >/dev/null 2>&1 && gcloud storage --help >/dev/null 2>&1; then
+  GCS_CMD="gcloud storage"
+fi
+readonly GCS_CMD
+
+function gcs_exists() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage objects describe "${uri}" >/dev/null 2>&1
+  else
+    gsutil -q stat "${uri}"
+  fi
+}
+
+function gcs_file_epoch() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    local file_time=$(gcloud storage objects describe "${uri}" --format="value(timeCreated)" 2>/dev/null || echo "")
+    if [[ -n "${file_time}" ]]; then
+      date -u -d "${file_time}" +%s 2>/dev/null || echo ""
+    fi
+  else
+    local file_time=$(gsutil stat "${uri}" | grep -oP 'Creation time:\s*\K.*' || echo "")
+    if [[ -n "${file_time}" ]]; then
+      date -u -d "${file_time}" +%s 2>/dev/null || echo ""
+    fi
+  fi
+}
+
+function gcs_cat() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cat "${uri}"
+  else
+    gsutil cat "${uri}"
+  fi
+}
+
+function gcs_atomic_upload() {
+  local local_path=$1
+  local gcs_uri=$2
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cp --if-generation-match=0 "${local_path}" "${gcs_uri}" 2>/dev/null
+  else
+    gsutil -h "x-goog-if-generation-match:0" cp "${local_path}" "${gcs_uri}" 2>/dev/null
+  fi
+}
+
+function gcs_cp() {
+  local local_path=$1
+  local gcs_uri=$2
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cp "${local_path}" "${gcs_uri}"
+  else
+    gsutil cp "${local_path}" "${gcs_uri}"
+  fi
+}
+
+function gcs_rm() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage rm "${uri}"
+  else
+    gsutil rm "${uri}"
+  fi
+}
+
+# Detect dataproc image version from its various names
+if (! test -v DATAPROC_IMAGE_VERSION) && test -v DATAPROC_VERSION; then
+  DATAPROC_IMAGE_VERSION="${DATAPROC_VERSION}"
+fi
+
+
+function err() {
+  echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $*" >&2
+  exit 1
+}
+
+function get_cache_key() {
+  local sorted_packages=$(echo "${PACKAGES}" | tr ' ' '\n' | sort | tr '\n' ' ' | xargs)
+  local sorted_channels=$(echo "${CHANNELS}" | tr ' ' '\n' | sort | tr '\n' ' ' | xargs)
+  # Only hash the sorted packages and channels, environment details are in the path
+  echo "${sorted_packages}_${sorted_channels}" | sha256sum | awk '{print $1}'
+}
+
+function main() {
+  if [[ -z "${PACKAGES}" ]]; then
+    echo "ERROR: Must specify CONDA_PACKAGES metadata key"
+    exit 1
+  fi
+
+  if [[ -z "${CACHE_BUCKET}" ]]; then
+    echo "ERROR: Must specify CACHE_BUCKET metadata key for cached installation"
+    exit 1
+  fi
+
+  local conda_path
+  conda_path=$(which conda 2>/dev/null) || true
+  local env_path
+
+  if [[ -n "${TARGET_ENV_PATH}" ]]; then
+    env_path="${TARGET_ENV_PATH}"
+    echo "Using target environment path from metadata: ${env_path}"
+  else
+    if [[ -z "${conda_path}" ]]; then
+      err "conda command not found and TARGET_ENV_PATH not specified."
+    fi
+    env_path=$(dirname $(dirname "${conda_path}"))
+    if [[ "${env_path}" == "/usr" || "${env_path}" == "/usr/local" || "${env_path}" == "/" || "${env_path}" == "/opt/conda" ]]; then
+      echo "WARNING: Inferred system-wide environment path: ${env_path}."
+      echo "Caching the entire system directory is unsafe and may result in large cache files. It is strongly recommended to specify a target environment path using TARGET_ENV_PATH."
+    fi
+    echo "Inferred target environment path: ${env_path}"
+  fi
+
+  echo "Target environment path: ${env_path}"
+  mkdir -p "${env_path}"
+
+  # Configure channels
+  if [[ -n "${CHANNELS}" ]]; then
+    if [[ -z "${conda_path}" ]]; then
+       echo "WARNING: Cannot configure channels, conda binary not found in PATH"
+    else
+      local -a chan_arr
+      read -r -a chan_arr <<< "${CHANNELS}"
+      for channel in "${chan_arr[@]}"; do
+        conda config --add channels "${channel}"
+      done
+    fi
+  fi
+
+
+  local cache_key
+  if [[ -n "${CACHE_KEY_OVERRIDE}" ]]; then
+    cache_key="${CACHE_KEY_OVERRIDE}"
+    echo "Using manual cache key override: ${cache_key}"
+  else
+    cache_key=$(get_cache_key)
+  fi
+
+  local arch=$(uname -m)
+  # Follow design doc storage layout: gs://<CACHE_BUCKET>/dataproc-caching/conda/<OS_DATAPROC_VER>_<ARCH>/<HASH_OR_KEY>.tar.gz
+  local gcs_tarball="gs://${CACHE_BUCKET}/dataproc-caching/conda/${OS_NAME}_${DATAPROC_IMAGE_VERSION}_${arch}/${cache_key}.tar.gz"
+  local local_tarball="/tmp/${cache_key}.tar.gz"
+
+  set +e
+  gcs_exists "${gcs_tarball}"
+  local cache_exists_code=$?
+  set -e
+
+  if [[ ${cache_exists_code} -eq 0 ]]; then
+    local file_epoch=$(gcs_file_epoch "${gcs_tarball}")
+
+    if [[ -n "${file_epoch}" ]]; then
+      local now_epoch=$(date -u +%s)
+      local age=$((now_epoch - file_epoch))
+
+      if (( age <= CACHE_TIMEOUT )); then
+        echo "Cache hit for key ${cache_key} (age: ${age}s). Unpacking from ${gcs_tarball}"
+        if gcs_cat "${gcs_tarball}" | tar -C "${env_path}" -xz; then
+          echo "Cache unpacked successfully."
+          return 0
+        else
+          echo "WARNING: Failed to unpack cache. Falling back to rebuilding."
+        fi
+      else
+        echo "Cache expired for key ${cache_key} (age: ${age}s > ${CACHE_TIMEOUT}s). Rebuilding."
+      fi
+    else
+      echo "Cache hit for key ${cache_key} (could not verify age). Unpacking."
+      if gcs_cat "${gcs_tarball}" | tar -C "${env_path}" -xz; then
+        return 0
+      else
+        echo "WARNING: Failed to unpack cache. Falling back to rebuilding."
+      fi
+    fi
+  fi
+
+  echo "Cache miss for key ${cache_key}. Checking for concurrent builds."
+
+  local -a pack_arr
+  read -r -a pack_arr <<< "${PACKAGES}"
+
+  while true; do
+    set +e
+    gcs_exists "${gcs_tarball}.building"
+    local sentinel_exists_code=$?
+    set -e
+
+    if [[ ${sentinel_exists_code} -eq 0 ]]; then
+      local sentinel_epoch=$(gcs_file_epoch "${gcs_tarball}.building")
+      if [[ -n "${sentinel_epoch}" ]]; then
+        local now_epoch=$(date -u +%s)
+        local sentinel_age=$((now_epoch - sentinel_epoch))
+        if (( sentinel_age > 1200 )); then
+          echo "Found stale sentinel (age: ${sentinel_age}s > 1200s). Removing it."
+          gcs_rm "${gcs_tarball}.building" || true
+          sentinel_exists_code=1
+        fi
+      fi
+    fi
+
+    if [[ ${sentinel_exists_code} -eq 0 ]]; then
+      echo "Another node is building this environment. Waiting..."
+      local wait_start=$(date +%s)
+      local timeout=1200
+      while gcs_exists "${gcs_tarball}.building"; do
+        if gcs_exists "${gcs_tarball}"; then
+          echo "Cache file appeared while waiting. Skipping build."
+          if gcs_cat "${gcs_tarball}" | tar -C "${env_path}" -xz; then
+            return 0
+          fi
+          echo "WARNING: Failed to unpack the newly available cache. Proceeding to build."
+          break
+        fi
+        local now=$(date +%s)
+        if (( now - wait_start > timeout )); then
+          echo "Timeout waiting for concurrent build. Removing stale sentinel and proceeding to build myself."
+          gcs_rm "${gcs_tarball}.building" || true
+          break
+        fi
+        echo "Waiting 30 seconds..."
+        sleep 30
+      done
+
+      # Double check if cache appeared just after sentinel removal
+      if gcs_exists "${gcs_tarball}"; then
+        echo "Cache file appeared. Skipping build."
+        if gcs_cat "${gcs_tarball}" | tar -C "${env_path}" -xz; then
+          return 0
+        fi
+        echo "WARNING: Failed to unpack the newly available cache. Proceeding to build."
+      fi
+    fi
+
+    echo "Proceeding to build and cache environment."
+
+    touch "${local_tarball}.building"
+
+    set +e
+    gcs_atomic_upload "${local_tarball}.building" "${gcs_tarball}.building"
+    local sentinel_upload_code=$?
+    set -e
+
+    if [[ ${sentinel_upload_code} -eq 0 ]]; then
+      break
+    fi
+
+    # Check if the failure was due to a race condition (sentinel exists) or GCS being unreachable
+    if ! gcs_exists "${gcs_tarball}.building"; then
+      echo "WARNING: Failed to upload sentinel to GCS. GCS might be unreachable."
+      echo "Falling back to standard non-cached installation."
+      conda install -y -p "${env_path}" "${pack_arr[@]}"
+      rm -f "${local_tarball}.building"
+      return 0
+    fi
+
+    echo "Lost the lock race to another node. Retrying..."
+  done
+
+  # Ensure we clean up sentinel on exit if we uploaded it
+  trap 'gcs_rm "${gcs_tarball}.building" || true; rm -f "${local_tarball}.building"' EXIT
+
+  # Perform standard installation
+  conda install -y -p "${env_path}" "${pack_arr[@]}"
+
+  echo "Packaging environment."
+  # We package the contents of the environment path
+  pushd "${env_path}"
+  tar czf "${local_tarball}" .
+  popd
+
+  echo "Uploading to GCS."
+  set +e
+  gcs_cp "${local_tarball}" "${gcs_tarball}"
+  local cache_upload_code=$?
+  set -e
+
+  if [[ ${cache_upload_code} -ne 0 ]]; then
+    echo "WARNING: Failed to upload cache to GCS."
+  fi
+
+  echo "Cleaning up."
+  trap - EXIT
+  gcs_rm "${gcs_tarball}.building" || true
+  rm -f "${local_tarball}" "${local_tarball}.building"
+
+  echo "Done."
+}
+
+main
diff --git a/python/pip-install-cached.sh b/python/pip-install-cached.sh
new file mode 100755
index 000000000..c9af9c057
--- /dev/null
+++ b/python/pip-install-cached.sh
@@ -0,0 +1,439 @@
+#!/bin/bash
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS-IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This script installs Python packages via pip, using GCS to cache the resulting environment.
+
+set -exo pipefail
+
+readonly PACKAGES=$(/usr/share/google/get_metadata_value attributes/PIP_PACKAGES || true)
+CACHE_BUCKET=$(/usr/share/google/get_metadata_value attributes/CACHE_BUCKET || true)
+if [[ -z "${CACHE_BUCKET}" ]]; then
+  CACHE_BUCKET=$(/usr/share/google/get_metadata_value attributes/dataproc-temp-bucket || true)
+  if [[ -n "${CACHE_BUCKET}" ]]; then
+    echo "CACHE_BUCKET not specified, defaulting to dataproc-temp-bucket: ${CACHE_BUCKET}"
+  fi
+fi
+readonly CACHE_BUCKET
+readonly CACHE_KEY_OVERRIDE=$(/usr/share/google/get_metadata_value attributes/CACHE_KEY_OVERRIDE || true)
+CACHE_TIMEOUT=$(/usr/share/google/get_metadata_value attributes/CACHE_TIMEOUT || true)
+if [[ -z "${CACHE_TIMEOUT}" ]]; then
+  CACHE_TIMEOUT=315360000 # 10 years (effectively infinite)
+fi
+readonly CACHE_TIMEOUT
+readonly TARGET_ENV_PATH=$(/usr/share/google/get_metadata_value attributes/TARGET_ENV_PATH || true)
+readonly OS_NAME=$(grep '^ID=' /etc/os-release | cut -d= -f2 | xargs | tr '[:upper:]' '[:lower:]')
+
+GCS_CMD="gsutil"
+if gcloud --help >/dev/null 2>&1 && gcloud storage --help >/dev/null 2>&1; then
+  GCS_CMD="gcloud storage"
+fi
+readonly GCS_CMD
+
+function gcs_exists() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage objects describe "${uri}" >/dev/null 2>&1
+  else
+    gsutil -q stat "${uri}"
+  fi
+}
+
+function gcs_file_epoch() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    local file_time=$(gcloud storage objects describe "${uri}" --format="value(timeCreated)" 2>/dev/null || echo "")
+    if [[ -n "${file_time}" ]]; then
+      date -u -d "${file_time}" +%s 2>/dev/null || echo ""
+    fi
+  else
+    local file_time=$(gsutil stat "${uri}" | grep -oP 'Creation time:\s*\K.*' || echo "")
+    if [[ -n "${file_time}" ]]; then
+      date -u -d "${file_time}" +%s 2>/dev/null || echo ""
+    fi
+  fi
+}
+
+function gcs_cat() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cat "${uri}"
+  else
+    gsutil cat "${uri}"
+  fi
+}
+
+function gcs_atomic_upload() {
+  local local_path=$1
+  local gcs_uri=$2
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cp --if-generation-match=0 "${local_path}" "${gcs_uri}" 2>/dev/null
+  else
+    gsutil -h "x-goog-if-generation-match:0" cp "${local_path}" "${gcs_uri}" 2>/dev/null
+  fi
+}
+
+function gcs_cp() {
+  local local_path=$1
+  local gcs_uri=$2
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage cp "${local_path}" "${gcs_uri}"
+  else
+    gsutil cp "${local_path}" "${gcs_uri}"
+  fi
+}
+
+function gcs_rm() {
+  local uri=$1
+  if [[ "${GCS_CMD}" == "gcloud storage" ]]; then
+    gcloud storage rm "${uri}"
+  else
+    gsutil rm "${uri}"
+  fi
+}
+
+# Detect dataproc image version from its various names
+if (! test -v DATAPROC_IMAGE_VERSION) && test -v DATAPROC_VERSION; then
+  DATAPROC_IMAGE_VERSION="${DATAPROC_VERSION}"
+fi
+
+
+function err() {
+  echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $*" >&2
+  exit 1
+}
+
+function remove_old_backports() {
+  # This script uses 'apt-get update' and is therefore potentially dependent on
+  # backports repositories which have been archived.  In order to mitigate this
+  # problem, we will remove any reference to backports repos older than oldstable
+
+  # https://github.com/GoogleCloudDataproc/initialization-actions/issues/1157
+  local oldstable
+  oldstable=$(curl -s --connect-timeout 5 --max-time 10 https://deb.debian.org/debian/dists/oldstable/Release 2>/dev/null | awk '/^Codename/ {print $2}' || echo "oldstable")
+  local stable
+  stable=$(curl -s --connect-timeout 5 --max-time 10 https://deb.debian.org/debian/dists/stable/Release 2>/dev/null | awk '/^Codename/ {print $2}' || echo "stable")
+
+
+  local matched_files=( $(grep -rsil '\-backports' /etc/apt/sources.list*||:) )
+  if [[ -n "$matched_files" ]]; then
+    for filename in "${matched_files[@]}"; do
+      grep -e "$oldstable-backports" -e "$stable-backports" "$filename" || \
+        sed -i -e 's/^.*-backports.*$//' "$filename"
+    done
+  fi
+}
+
+function run_with_retry() {
+  local -r cmd=("$@")
+  for ((i = 0; i < 10; i++)); do
+    if "${cmd[@]}"; then
+      return 0
+    fi
+    sleep 5
+  done
+  err "Failed to run command: ${cmd[*]}"
+}
+
+function install_pip() {
+  if command -v pip >/dev/null; then
+    echo "pip is already installed."
+    return 0
+  fi
+
+  if command -v easy_install >/dev/null; then
+    echo "Installing pip with easy_install..."
+    run_with_retry easy_install pip
+    return 0
+  fi
+
+  echo "Installing python3-pip..."
+  run_with_retry apt update
+  run_with_retry apt install python3-pip -y
+}
+
+function get_cache_key() {
+  local sorted_packages=$(echo "${PACKAGES}" | tr ' ' '\n' | sort | tr '\n' ' ' | xargs)
+  # Only hash the sorted packages, environment details are in the path
+  echo "${sorted_packages}" | sha256sum | awk '{print $1}'
+}
+
+function ensure_venv() {
+  if python3 -m venv --help >/dev/null 2>&1; then
+    echo "venv is available."
+    return 0
+  fi
+
+  echo "Installing venv..."
+  if [[ "${OS_NAME}" == "debian" || "${OS_NAME}" == "ubuntu" ]]; then
+    run_with_retry apt update
+    run_with_retry apt install python3-venv -y
+  elif [[ "${OS_NAME}" == "rocky" ]]; then
+    run_with_retry yum install -y python3-virtualenv || run_with_retry dnf install -y python3-virtualenv
+  fi
+}
+
+function merge_isolated_env() {
+  local venv_path=$1
+  echo "Merging isolated environment back to system Python"
+  local system_site_packages=$(/usr/bin/python3 -c "import site; print(site.getsitepackages()[0])" 2>/dev/null)
+  local venv_site_packages=$(find "${venv_path}/lib" -maxdepth 2 -type d -name "site-packages" 2>/dev/null | head -n 1)
+  if [[ -n "${system_site_packages}" && -n "${venv_site_packages}" ]]; then
+    cp -a "${venv_site_packages}/." "${system_site_packages}/"
+    if [[ -d "${venv_path}/bin" ]]; then
+      find "${venv_path}/bin" -type f ! -name "python*" ! -name "pip*" ! -name "easy_install*" ! -name "activate*" -exec cp -a {} /usr/local/bin/ \;
+      # Rewrite shebangs of copied scripts to use system python
+      for f in $(find "${venv_path}/bin" -type f ! -name "python*" ! -name "pip*" ! -name "easy_install*" ! -name "activate*" -exec basename {} \;); do
+        if head -n 1 "/usr/local/bin/${f}" | grep -q "python"; then
+          sed -i "1s|^#!.*python.*|#!/usr/bin/env python3|" "/usr/local/bin/${f}"
+        fi
+      done
+    fi
+  else
+    echo "ERROR: Could not find system or venv site-packages directory"
+    return 1
+  fi
+}
+
+function unpack_and_merge() {
+  local tarball=$1
+  local target_path=$2
+  local original_path=$3
+  local is_isolated=$4
+
+  mkdir -p "${target_path}"
+  if gcs_cat "${tarball}" | tar -C "${target_path}" -xz; then
+    if [[ "${is_isolated}" == "true" ]]; then
+      merge_isolated_env "${target_path}" || return 1
+    fi
+    return 0
+  fi
+  return 1
+}
+
+function main() {
+  if [[ -z "${PACKAGES}" ]]; then
+    echo "ERROR: Must specify PIP_PACKAGES metadata key"
+    exit 1
+  fi
+
+  if [[ -z "${CACHE_BUCKET}" ]]; then
+    echo "ERROR: Must specify CACHE_BUCKET metadata key for cached installation"
+    exit 1
+  fi
+
+  local env_path
+  local pip_bin
+  local original_env_path=""
+  local isolated_env_path="/opt/pip-isolated-env"
+  local isolated="false"
+
+  if [[ -n "${TARGET_ENV_PATH}" ]]; then
+    env_path="${TARGET_ENV_PATH}"
+    pip_bin="${env_path}/bin/pip"
+    echo "Using target environment path from metadata: ${env_path}"
+  else
+    if [[ "${OS_NAME}" == "debian" ]] && [[ -n "${DATAPROC_IMAGE_VERSION}" ]]; then
+      local dp_ver=$(echo "${DATAPROC_IMAGE_VERSION}" | sed -E 's/^([0-9]+\.[0-9]+).*$/\1/')
+      if [[ $(echo "${dp_ver} <= 2.1" | bc -l 2>/dev/null) == 1 ]]; then
+        remove_old_backports
+      fi
+    fi
+    install_pip
+    local pip_path=$(which pip)
+    env_path=$(dirname $(dirname "${pip_path}"))
+    if [[ "${env_path}" == "/usr" || "${env_path}" == "/usr/local" || "${env_path}" == "/" ]]; then
+      echo "WARNING: Inferred system-wide environment path: ${env_path}."
+      echo "Switching to isolated venv for caching to avoid archiving system directories."
+      ensure_venv
+      python3 -m venv --system-site-packages "${isolated_env_path}" || virtualenv --system-site-packages "${isolated_env_path}"
+
+      original_env_path="${env_path}"
+      env_path="${isolated_env_path}"
+      pip_bin="${env_path}/bin/pip"
+      isolated="true"
+    else
+      pip_bin="pip"
+    fi
+    echo "Inferred target environment path: ${env_path}"
+  fi
+
+  echo "Target environment path: ${env_path}"
+
+  local cache_key
+  if [[ -n "${CACHE_KEY_OVERRIDE}" ]]; then
+    cache_key="${CACHE_KEY_OVERRIDE}"
+    echo "Using manual cache key override: ${cache_key}"
+  else
+    cache_key=$(get_cache_key)
+  fi
+
+  local arch=$(uname -m)
+  # Follow design doc storage layout: gs://<CACHE_BUCKET>/dataproc-caching/pip/<OS_DATAPROC_VER>_<ARCH>/<HASH_OR_KEY>.tar.gz
+  local gcs_tarball="gs://${CACHE_BUCKET}/dataproc-caching/pip/${OS_NAME}_${DATAPROC_IMAGE_VERSION}_${arch}/${cache_key}.tar.gz"
+  local local_tarball="/tmp/${cache_key}.tar.gz"
+
+  set +e
+  gcs_exists "${gcs_tarball}"
+  local cache_exists_code=$?
+  set -e
+
+  if [[ ${cache_exists_code} -eq 0 ]]; then
+    local file_epoch=$(gcs_file_epoch "${gcs_tarball}")
+
+    if [[ -n "${file_epoch}" ]]; then
+      local now_epoch=$(date -u +%s)
+      local age=$((now_epoch - file_epoch))
+
+      if (( age <= CACHE_TIMEOUT )); then
+        echo "Cache hit for key ${cache_key} (age: ${age}s). Unpacking from ${gcs_tarball}"
+        if unpack_and_merge "${gcs_tarball}" "${env_path}" "${original_env_path}" "${isolated}"; then
+          echo "Cache unpacked successfully."
+          return 0
+        else
+          echo "WARNING: Failed to unpack cache. Falling back to rebuilding."
+        fi
+      else
+        echo "Cache expired for key ${cache_key} (age: ${age}s > ${CACHE_TIMEOUT}s). Rebuilding."
+      fi
+    else
+      echo "Cache hit for key ${cache_key} (could not verify age). Unpacking."
+      if unpack_and_merge "${gcs_tarball}" "${env_path}" "${original_env_path}" "${isolated}"; then
+        return 0
+      else
+        echo "WARNING: Failed to unpack cache. Falling back to rebuilding."
+      fi
+    fi
+  fi
+
+  echo "Cache miss for key ${cache_key}. Checking for concurrent builds."
+
+  while true; do
+    set +e
+    gcs_exists "${gcs_tarball}.building"
+    local sentinel_exists_code=$?
+    set -e
+
+    if [[ ${sentinel_exists_code} -eq 0 ]]; then
+      local sentinel_epoch=$(gcs_file_epoch "${gcs_tarball}.building")
+      if [[ -n "${sentinel_epoch}" ]]; then
+        local now_epoch=$(date -u +%s)
+        local sentinel_age=$((now_epoch - sentinel_epoch))
+        if (( sentinel_age > 1200 )); then
+          echo "Found stale sentinel (age: ${sentinel_age}s > 1200s). Removing it."
+          gcs_rm "${gcs_tarball}.building" || true
+          sentinel_exists_code=1
+        fi
+      fi
+    fi
+
+    if [[ ${sentinel_exists_code} -eq 0 ]]; then
+      echo "Another node is building this environment. Waiting..."
+      local wait_start=$(date +%s)
+      local timeout=1200
+      while gcs_exists "${gcs_tarball}.building"; do
+        if gcs_exists "${gcs_tarball}"; then
+          echo "Cache file appeared while waiting. Skipping build."
+          if unpack_and_merge "${gcs_tarball}" "${env_path}" "${original_env_path}" "${isolated}"; then
+            return 0
+          fi
+          echo "WARNING: Failed to unpack the newly available cache. Proceeding to build."
+          break
+        fi
+        local now=$(date +%s)
+        if (( now - wait_start > timeout )); then
+          echo "Timeout waiting for concurrent build. Removing stale sentinel and proceeding to build myself."
+          gcs_rm "${gcs_tarball}.building" || true
+          break
+        fi
+        echo "Waiting 30 seconds..."
+        sleep 30
+      done
+      
+      # Double check if cache appeared just after sentinel removal
+      if gcs_exists "${gcs_tarball}"; then
+        echo "Cache file appeared. Skipping build."
+        if unpack_and_merge "${gcs_tarball}" "${env_path}" "${original_env_path}" "${isolated}"; then
+          return 0
+        fi
+        echo "WARNING: Failed to unpack the newly available cache. Proceeding to build."
+      fi
+    fi
+
+    echo "Proceeding to build and cache environment."
+    
+    touch "${local_tarball}.building"
+    
+    set +e
+    gcs_atomic_upload "${local_tarball}.building" "${gcs_tarball}.building"
+    local sentinel_upload_code=$?
+    set -e
+
+    if [[ ${sentinel_upload_code} -eq 0 ]]; then
+      break
+    fi
+
+    # Check if the failure was due to a race condition (sentinel exists) or GCS being unreachable
+    if ! gcs_exists "${gcs_tarball}.building"; then
+      echo "WARNING: Failed to upload sentinel to GCS. GCS might be unreachable."
+      echo "Falling back to standard non-cached installation."
+      local -a pack_arr
+      read -r -a pack_arr <<< "${PACKAGES}"
+      run_with_retry "${pip_bin}" install --upgrade "${pack_arr[@]}"
+      if [[ "${isolated}" == "true" ]]; then
+        merge_isolated_env "${env_path}"
+      fi
+      rm -f "${local_tarball}.building"
+      return 0
+    fi
+
+    echo "Lost the lock race to another node. Retrying..."
+  done
+
+  # Ensure we clean up sentinel on exit if we uploaded it
+  trap 'gcs_rm "${gcs_tarball}.building" || true; rm -f "${local_tarball}.building"' EXIT
+
+  # Perform standard installation
+  local -a pack_arr
+  read -r -a pack_arr <<< "${PACKAGES}"
+  run_with_retry "${pip_bin}" install --upgrade "${pack_arr[@]}"
+
+  if [[ "${isolated}" == "true" ]]; then
+    merge_isolated_env "${env_path}"
+  fi
+
+  echo "Packaging environment."
+  # We package the contents of the environment path
+  pushd "${env_path}"
+  tar czf "${local_tarball}" .
+  popd
+
+  echo "Uploading to GCS."
+  set +e
+  gcs_cp "${local_tarball}" "${gcs_tarball}"
+  local cache_upload_code=$?
+  set -e
+
+  if [[ ${cache_upload_code} -ne 0 ]]; then
+    echo "WARNING: Failed to upload cache to GCS."
+  fi
+
+  echo "Cleaning up."
+  trap - EXIT
+  gcs_rm "${gcs_tarball}.building" || true
+  rm -f "${local_tarball}" "${local_tarball}.building"
+
+  echo "Done."
+}
+
+main
diff --git a/python/test_conda_cached.py b/python/test_conda_cached.py
new file mode 100644
index 000000000..039be51b0
--- /dev/null
+++ b/python/test_conda_cached.py
@@ -0,0 +1,65 @@
+import json
+import packaging.version
+from absl.testing import absltest
+from absl.testing import parameterized
+
+from integration_tests.dataproc_test_case import DataprocTestCase
+
+CONDA_BINARY = "/opt/conda/bin/conda"
+
+class CondaCachedTestCase(DataprocTestCase):
+    COMPONENT = "python"
+    INIT_ACTIONS = ["python/conda-install-cached.sh"]
+
+    # Test packages
+    CONDA_PKGS = ["numpy"]
+    GCS_BUCKET = None
+
+    def setUp(self):
+        super().setUp()
+        self.GCS_BUCKET = "test-conda-cached-{}-{}".format(self.datetime_str(),
+                                                           self.random_str())
+        self.assert_command('gsutil mb -c regional -l {} gs://{}'.format(
+            self.REGION, self.GCS_BUCKET))
+
+    def tearDown(self):
+        if self.GCS_BUCKET:
+            self.run_command('gsutil -m rm -rf gs://{}'.format(self.GCS_BUCKET))
+        super().tearDown()
+
+    def _verify_conda_packages(self, instance, conda_packages):
+        _, stdout, _ = self.assert_instance_command(instance, CONDA_BINARY + " list")
+        installed_packages = self._parse_packages(stdout)
+        for package in conda_packages:
+            self.assertIn(
+                package, installed_packages,
+                "Expected package {} to be installed, but wasn't."
+                " Packages installed: {}".format(package, installed_packages))
+
+    def _verify_cache_created(self):
+        ret_code, stdout, _ = self.run_command("gsutil ls gs://{}".format(self.GCS_BUCKET))
+        self.assertEqual(ret_code, 0, "Failed to list bucket")
+        self.assertTrue(len(stdout.strip()) > 0, "Cache was not created in GCS")
+
+    @staticmethod
+    def _parse_packages(stdout):
+        return set(
+            l.split()[0] for l in stdout.splitlines() if not l.startswith("#"))
+
+    @parameterized.parameters(
+        ("STANDARD", CONDA_PKGS),
+    )
+    def test_conda_cached(self, configuration, conda_packages):
+        metadata = "'CONDA_PACKAGES={},CACHE_BUCKET={}'".format(
+            " ".join(conda_packages), self.GCS_BUCKET)
+        self.createCluster(
+            configuration,
+            self.INIT_ACTIONS,
+            metadata=metadata)
+
+        instance_name = self.getClusterName() + "-m"
+        self._verify_conda_packages(instance_name, conda_packages)
+        self._verify_cache_created()
+
+if __name__ == "__main__":
+    absltest.main()
diff --git a/python/test_pip_cached.py b/python/test_pip_cached.py
new file mode 100644
index 000000000..5ec5ba97a
--- /dev/null
+++ b/python/test_pip_cached.py
@@ -0,0 +1,70 @@
+import json
+import packaging.version
+from absl.testing import absltest
+from absl.testing import parameterized
+
+from integration_tests.dataproc_test_case import DataprocTestCase
+
+class PipCachedTestCase(DataprocTestCase):
+    COMPONENT = "python"
+    INIT_ACTIONS = ["python/pip-install-cached.sh"]
+
+    # Test packages
+    PIP_PKGS = ["pandas-gbq"]
+    GCS_BUCKET = None
+
+    def setUp(self):
+        super().setUp()
+        self.GCS_BUCKET = "test-pip-cached-{}-{}".format(self.datetime_str(),
+                                                         self.random_str())
+        self.assert_command('gsutil mb -c regional -l {} gs://{}'.format(
+            self.REGION, self.GCS_BUCKET))
+
+    def tearDown(self):
+        if self.GCS_BUCKET:
+            self.run_command('gsutil -m rm -rf gs://{}'.format(self.GCS_BUCKET))
+        super().tearDown()
+
+    def _verify_pip_packages(self, instance, pip_packages):
+        # Try both pip and python3 -m pip to be robust
+        _, stdout, _ = self.assert_instance_command(instance, "python3 -m pip list")
+        installed_packages = self._parse_packages(stdout)
+        for package in pip_packages:
+            if package not in installed_packages:
+                # Try pip list as fallback
+                _, stdout_pip, _ = self.assert_instance_command(instance, "pip list")
+                installed_packages_pip = self._parse_packages(stdout_pip)
+                self.assertIn(
+                    package, installed_packages_pip,
+                    "Expected package {} to be installed, but wasn't."
+                    " Packages installed (python3 -m pip): {}"
+                    " Packages installed (pip): {}".format(
+                        package, installed_packages, installed_packages_pip))
+
+    def _verify_cache_created(self):
+        ret_code, stdout, _ = self.run_command("gsutil ls gs://{}".format(self.GCS_BUCKET))
+        self.assertEqual(ret_code, 0, "Failed to list bucket")
+        self.assertTrue(len(stdout.strip()) > 0, "Cache was not created in GCS")
+
+    @staticmethod
+    def _parse_packages(stdout):
+        return set(
+            l.split()[0] for l in stdout.splitlines() if not l.startswith("#"))
+
+    @parameterized.parameters(
+        ("STANDARD", PIP_PKGS),
+    )
+    def test_pip_cached(self, configuration, pip_packages):
+        metadata = "'PIP_PACKAGES={},CACHE_BUCKET={}'".format(
+            " ".join(pip_packages), self.GCS_BUCKET)
+        self.createCluster(
+            configuration,
+            self.INIT_ACTIONS,
+            metadata=metadata)
+
+        instance_name = self.getClusterName() + "-m"
+        self._verify_pip_packages(instance_name, pip_packages)
+        self._verify_cache_created()
+
+if __name__ == "__main__":
+    absltest.main()