Move GCP args from DataprocSubmitter to CLI

api/python/ai/chronon/repo/gcp.py

@@ -298,14 +298,24 @@ def generate_dataproc_submitter_args(
 
         elif job_type == JobType.SPARK:
             main_class = "ai.chronon.spark.Driver"
-            return final_args.format(
-                user_args=user_args,
-                jar_uri=jar_uri,
-                job_type=job_type.value,
-                main_class=main_class,
-                zipline_version=self._version,
-                job_id=self.job_id,
-            ) + (f" --files={gcs_file_args}" if gcs_file_args else "")
+            return (
+                final_args.format(
+                    user_args=user_args,
+                    jar_uri=jar_uri,
+                    job_type=job_type.value,
+                    main_class=main_class,
+                    zipline_version=self._version,
+                    job_id=self.job_id,
+                )
+                + " ".join(
+                    [
+                        "--is-gcp",
+                        f"--gcp-project-id={GcpRunner.get_gcp_project_id()}",
+                        f"--gcp-bigtable-instance-id={GcpRunner.get_gcp_bigtable_instance_id()}--",
+                        f" --files={gcs_file_args}" if gcs_file_args else "",
+                    ]
+                )
+            )
         else:
             raise ValueError(f"Invalid job type: {job_type}")
 

cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala

@@ -440,6 +440,8 @@ object DataprocSubmitter {
       case TypeSparkJob => {
         val bigtableInstanceId = envMap(GcpBigtableInstanceIdEnvVar).getOrElse("")
         val projectId = envMap(GcpProjectIdEnvVar).getOrElse(throw new Exception(s"GcpProjectId not set"))
+
+        // TODO: remove gcpArgsToPass and have it be passed from the old python CLI
         val gcpArgsToPass = Array.apply(
           "--is-gcp",
           s"--gcp-project-id=${projectId}",

spark/src/main/scala/ai/chronon/spark/submission/JobSubmitter.scala

@@ -172,6 +172,7 @@ object JobSubmitterConstants {
   val AdditionalJarsUriArgKeyword = "--additional-jars"
   val FilesArgKeyword = "--files"
   val ConfTypeArgKeyword = "--conf-type"
+  val ConfPathArgKeyword = "--conf-path"
   val LocalConfPathArgKeyword = "--local-conf-path"
   val OriginalModeArgKeyword = "--original-mode"
   val ZiplineVersionArgKeyword = "--zipline-version"