Skip to content

Commit 0c88774

Browse files
[CATALYST-262] Stage Stitch job JAR via Unity Catalog volume (#91)
## Summary Jira ticket: [CATALYST-262] Consumer-side wire-up for the chuck-api change in [CATALYST-253]. When the cluster init script lives in a Unity Catalog volume, we now generate a timestamped `JOB_JAR_VOL_PATH` co-located with it, export it in `spark_env_vars`, and switch the `Run_Stitch` task's library JAR from the local `file:///opt/amperity/job.jar` to that volume path. The chuck-api init script copies the downloaded JAR into `JOB_JAR_VOL_PATH`, and Databricks' synchronous library load blocks `Run_Stitch` until the JAR materializes there — no separate preflight task needed. ## Changes - `chuck_data/clients/databricks.py` - New `_generate_jar_volume_path(init_script_path)` helper returning `<volume-dir>/job-YYYYMMDD-HHMMSS.jar`. - `_build_libraries` accepts an optional `main_jar_path` override so the Run_Stitch library jar can point at the volume path. - `submit_job_run` populates `JOB_JAR_VOL_PATH` in `spark_env_vars` and passes the volume path to `_build_libraries` when the init script is a Volumes path. S3 init scripts (Redshift) keep the local `file:///` jar — there is no Unity Catalog volume to stage to. - `tests/unit/test_workspace_and_init_scripts.py` - New tests for the volumes path (timestamped jar + env var + library consistency) and the S3 path (no `JOB_JAR_VOL_PATH`, library jar unchanged). ## Testing - `python -m pytest tests/unit` — 1221 passed. ## Rollout Standard release. The change requires the matching chuck-api init script (CATALYST-253), which is already merged. [CATALYST-262]: https://amperity.atlassian.net/browse/CATALYST-262?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ [CATALYST-253]: https://amperity.atlassian.net/browse/CATALYST-253?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ
1 parent a61b92a commit 0c88774

2 files changed

Lines changed: 102 additions & 13 deletions

File tree

chuck_data/clients/databricks.py

Lines changed: 49 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
import requests
1010
import time
1111
import urllib.parse
12-
from datetime import datetime
12+
from datetime import datetime, timezone
1313
from chuck_data.config import get_warehouse_id
1414
from chuck_data.clients.amperity import get_amperity_url
1515
from chuck_data.databricks.url_utils import (
@@ -576,15 +576,38 @@ def submit_sql_statement(
576576
# Jobs methods
577577
#
578578

579-
def _build_libraries(self, data_provider=None):
579+
@staticmethod
580+
def _generate_jar_volume_path(init_script_path):
581+
"""Return a timestamped JAR volume path co-located with the init script.
582+
583+
Mirrors the chuck-api side (CATALYST-253): the cluster init script
584+
copies /opt/amperity/job.jar to JOB_JAR_VOL_PATH during cluster
585+
startup, and the Run_Stitch task's library entry points at this
586+
same path so Databricks blocks task start until the JAR is staged.
587+
The filename is timestamped so concurrent runs do not clobber each
588+
other's JARs.
589+
"""
590+
parent_dir = init_script_path.rsplit("/", 1)[0]
591+
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
592+
return f"{parent_dir}/job-{timestamp}.jar"
593+
594+
def _build_libraries(self, data_provider=None, main_jar_path=None):
580595
"""Return the Maven/JAR library list for a Stitch job run.
581596
582597
The main JAR is always included. The connector Maven dependency is chosen
583598
based on the data provider:
584599
- "snowflake" → Snowflake Spark connector
585600
- "aws_redshift" → Redshift community connector + Avro support
601+
602+
Args:
603+
data_provider: Optional connector selector (see above).
604+
main_jar_path: Optional override for the main JAR path. Defaults to
605+
the local `file:///opt/amperity/job.jar` path. Callers that
606+
stage the JAR into a Unity Catalog volume should pass that
607+
volume path so Databricks blocks the task until the init
608+
script has copied the JAR into the volume.
586609
"""
587-
libraries: list = [{"jar": "file:///opt/amperity/job.jar"}]
610+
libraries: list = [{"jar": main_jar_path or "file:///opt/amperity/job.jar"}]
588611
if data_provider == "snowflake":
589612
libraries.append(
590613
{"maven": {"coordinates": "net.snowflake:spark-snowflake_2.12:3.1.3"}}
@@ -634,9 +657,11 @@ def submit_job_run(
634657
# Define the task and cluster for the one-time run
635658
# Create base cluster configuration
636659
# Detect init script location (S3 vs Volumes) and configure accordingly
660+
jar_vol_path = None
637661
if init_script_path.startswith("s3://"):
638-
# S3 init script (for Redshift data source)
639-
# Get region from config
662+
# S3 init script (for Redshift data source). Volumes are a
663+
# Unity Catalog concept and don't apply here, so we leave the
664+
# library jar pointing at the local /opt/amperity/job.jar.
640665
from chuck_data.config import get_aws_region
641666

642667
region = get_aws_region() or "us-west-2"
@@ -650,14 +675,27 @@ def submit_job_run(
650675
}
651676
]
652677
else:
653-
# Volumes init script (for Unity Catalog data source)
678+
# Volumes init script (Unity Catalog). Stage the JAR into the
679+
# same volume so the Run_Stitch task's library entry resolves
680+
# against the volume path -- the chuck-api init script reads
681+
# JOB_JAR_VOL_PATH from spark_env_vars and copies the JAR there.
654682
init_scripts_config = [
655683
{
656684
"volumes": {
657685
"destination": init_script_path,
658686
}
659687
}
660688
]
689+
jar_vol_path = self._generate_jar_volume_path(init_script_path)
690+
691+
spark_env_vars = {
692+
"JNAME": "zulu17-ca-amd64",
693+
"CHUCK_API_URL": f"https://{get_amperity_url()}",
694+
"DEBUG_INIT_SRIPT_URL": init_script_path,
695+
"DEBUG_CONFIG_PATH": config_path,
696+
}
697+
if jar_vol_path:
698+
spark_env_vars["JOB_JAR_VOL_PATH"] = jar_vol_path
661699

662700
cluster_config = {
663701
"cluster_name": "",
@@ -669,12 +707,7 @@ def submit_job_run(
669707
"sys": "chuck",
670708
"tenant": "amperity",
671709
},
672-
"spark_env_vars": {
673-
"JNAME": "zulu17-ca-amd64",
674-
"CHUCK_API_URL": f"https://{get_amperity_url()}",
675-
"DEBUG_INIT_SRIPT_URL": init_script_path,
676-
"DEBUG_CONFIG_PATH": config_path,
677-
},
710+
"spark_env_vars": spark_env_vars,
678711
"enable_elastic_disk": False,
679712
"data_security_mode": "SINGLE_USER",
680713
"runtime_engine": "STANDARD",
@@ -704,7 +737,10 @@ def submit_job_run(
704737
],
705738
"run_as_repl": True,
706739
},
707-
"libraries": self._build_libraries(data_provider),
740+
"libraries": self._build_libraries(
741+
data_provider=data_provider,
742+
main_jar_path=jar_vol_path,
743+
),
708744
"timeout_seconds": 0,
709745
"email_notifications": {},
710746
"webhook_notifications": {},

tests/unit/test_workspace_and_init_scripts.py

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,59 @@ def test_submit_job_run_with_volumes_init_script(self, client):
112112

113113
assert result == {"run_id": 123}
114114

115+
def test_submit_job_run_volumes_stages_jar_into_volume(self, client):
116+
"""Volumes init script gets a timestamped JOB_JAR_VOL_PATH env var
117+
and the Run_Stitch library jar matches that path so Databricks
118+
blocks the task until the init script stages the JAR there."""
119+
import re
120+
121+
with patch.object(client, "post") as mock_post:
122+
mock_post.return_value = {"run_id": 321}
123+
124+
client.submit_job_run(
125+
config_path="/Volumes/cat/schema/vol/config.json",
126+
init_script_path="/Volumes/cat/schema/vol/init.sh",
127+
)
128+
129+
payload = mock_post.call_args[0][1]
130+
task = payload["tasks"][0]
131+
spark_env_vars = task["new_cluster"]["spark_env_vars"]
132+
133+
# JOB_JAR_VOL_PATH is a timestamped job-*.jar co-located with
134+
# the init script.
135+
jar_vol_path = spark_env_vars["JOB_JAR_VOL_PATH"]
136+
assert re.fullmatch(
137+
r"/Volumes/cat/schema/vol/job-\d{8}-\d{6}\.jar", jar_vol_path
138+
)
139+
140+
# Library jar entry matches JOB_JAR_VOL_PATH exactly so the
141+
# cluster's library load waits for the init script to stage
142+
# the JAR.
143+
jar_entries = [lib["jar"] for lib in task["libraries"] if "jar" in lib]
144+
assert jar_entries == [jar_vol_path]
145+
146+
def test_submit_job_run_s3_does_not_stage_jar_into_volume(self, client):
147+
"""S3 init scripts (Redshift) keep the local file:// jar -- there
148+
is no Unity Catalog volume to stage to and JOB_JAR_VOL_PATH must
149+
not be set."""
150+
with patch("chuck_data.config.get_aws_region", return_value="us-east-1"):
151+
with patch.object(client, "post") as mock_post:
152+
mock_post.return_value = {"run_id": 654}
153+
154+
client.submit_job_run(
155+
config_path="s3://bucket/config.json",
156+
init_script_path="s3://bucket/init.sh",
157+
)
158+
159+
payload = mock_post.call_args[0][1]
160+
task = payload["tasks"][0]
161+
spark_env_vars = task["new_cluster"]["spark_env_vars"]
162+
163+
assert "JOB_JAR_VOL_PATH" not in spark_env_vars
164+
165+
jar_entries = [lib["jar"] for lib in task["libraries"] if "jar" in lib]
166+
assert jar_entries == ["file:///opt/amperity/job.jar"]
167+
115168
@patch("chuck_data.config.get_aws_region")
116169
def test_submit_job_run_with_s3_init_script(self, mock_get_region, client):
117170
"""Test submit_job_run uses s3 format for S3 paths."""

0 commit comments

Comments
 (0)