Support for Spark Submit added in Data Flow

Author: rashik-bhasin

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Added
 - Support for `OPSI` service
+- Support for Spark Submit added in Data Flow
 
 ## 4.26.0 (May 12, 2021)
 

examples/dataflow/main.tf

@@ -200,6 +200,34 @@ resource "oci_dataflow_invoke_run" "test_invoke_run" {
   display_name   = "test_run_name"
 }
 
+resource "oci_dataflow_application" "test_application_submit" {
+  #Required
+  compartment_id = var.compartment_id
+  execute        = "--conf spark.shuffle.io.maxRetries=10 ${var.application_file_uri} arguments"
+  display_name   = "test_wordcount_app_submit"
+  driver_shape   = "VM.Standard2.1"
+  executor_shape = "VM.Standard2.1"
+  file_uri       = var.application_file_uri
+  language       = "PYTHON"
+  num_executors  = "1"
+  spark_version  = "2.4"
+  #Optional
+  archive_uri    = var.application_archive_uri
+  private_endpoint_id = oci_dataflow_private_endpoint.test_private_endpoint.id
+
+}
+
+resource "oci_dataflow_invoke_run" "test_invokey_run_submit" {
+  #Required
+  compartment_id = var.compartment_id
+  execute        = "--conf spark.shuffle.io.maxRetries=10 ${var.application_file_uri} arguments"
+  #Optional
+  application_id = oci_dataflow_application.test_application_submit.id
+  archive_uri    = var.application_archive_uri
+  display_name   = "test_wordcount_run_submit"
+  spark_version  = "2.4"
+}
+
 data "oci_dataflow_private_endpoints" "test_private_endpoints" {
   compartment_id = var.compartment_id
 

oci/dataflow_application_data_source.go

@@ -99,6 +99,10 @@ func (s *DataflowApplicationDataSourceCrud) SetData() error {
 		s.D.Set("driver_shape", *s.Res.DriverShape)
 	}
 
+	if s.Res.Execute != nil {
+		s.D.Set("execute", *s.Res.Execute)
+	}
+
 	if s.Res.ExecutorShape != nil {
 		s.D.Set("executor_shape", *s.Res.ExecutorShape)
 	}

oci/dataflow_application_resource.go

@@ -96,6 +96,11 @@ func DataflowApplicationResource() *schema.Resource {
 				Optional: true,
 				Computed: true,
 			},
+			"execute": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+			},
 			"freeform_tags": {
 				Type:     schema.TypeMap,
 				Optional: true,
@@ -287,6 +292,11 @@ func (s *DataflowApplicationResourceCrud) Create() error {
 		request.DriverShape = &tmp
 	}
 
+	if execute, ok := s.D.GetOkExists("execute"); ok {
+		tmp := execute.(string)
+		request.Execute = &tmp
+	}
+
 	if executorShape, ok := s.D.GetOkExists("executor_shape"); ok {
 		tmp := executorShape.(string)
 		request.ExecutorShape = &tmp
@@ -440,6 +450,11 @@ func (s *DataflowApplicationResourceCrud) Update() error {
 		request.DriverShape = &tmp
 	}
 
+	if execute, ok := s.D.GetOkExists("execute"); ok {
+		tmp := execute.(string)
+		request.Execute = &tmp
+	}
+
 	if executorShape, ok := s.D.GetOkExists("executor_shape"); ok {
 		tmp := executorShape.(string)
 		request.ExecutorShape = &tmp
@@ -556,6 +571,10 @@ func (s *DataflowApplicationResourceCrud) SetData() error {
 		s.D.Set("driver_shape", *s.Res.DriverShape)
 	}
 
+	if s.Res.Execute != nil {
+		s.D.Set("execute", *s.Res.Execute)
+	}
+
 	if s.Res.ExecutorShape != nil {
 		s.D.Set("executor_shape", *s.Res.ExecutorShape)
 	}

oci/dataflow_application_submit_test.go

@@ -108,6 +108,10 @@ func (s *DataflowInvokeRunDataSourceCrud) SetData() error {
 		s.D.Set("driver_shape", *s.Res.DriverShape)
 	}
 
+	if s.Res.Execute != nil {
+		s.D.Set("execute", *s.Res.Execute)
+	}
+
 	if s.Res.ExecutorShape != nil {
 		s.D.Set("executor_shape", *s.Res.ExecutorShape)
 	}

oci/dataflow_invoke_run_data_source.go

@@ -29,22 +29,24 @@ func DataflowInvokeRunResource() *schema.Resource {
 		Delete:   deleteDataflowInvokeRun,
 		Schema: map[string]*schema.Schema{
 			// Required
-			"application_id": {
+			"compartment_id": {
 				Type:     schema.TypeString,
 				Required: true,
-				ForceNew: true,
 			},
-			"compartment_id": {
+
+			// Optional
+			"application_id": {
 				Type:     schema.TypeString,
-				Required: true,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
 			},
-			"display_name": {
+			"archive_uri": {
 				Type:     schema.TypeString,
-				Required: true,
+				Optional: true,
+				Computed: true,
 				ForceNew: true,
 			},
-
-			// Optional
 			"arguments": {
 				Type:     schema.TypeList,
 				Optional: true,
@@ -68,12 +70,24 @@ func DataflowInvokeRunResource() *schema.Resource {
 				DiffSuppressFunc: definedTagsDiffSuppressFunction,
 				Elem:             schema.TypeString,
 			},
+			"display_name": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
 			"driver_shape": {
 				Type:     schema.TypeString,
 				Optional: true,
 				Computed: true,
 				ForceNew: true,
 			},
+			"execute": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
 			"executor_shape": {
 				Type:     schema.TypeString,
 				Optional: true,
@@ -123,6 +137,12 @@ func DataflowInvokeRunResource() *schema.Resource {
 					},
 				},
 			},
+			"spark_version": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
 			"warehouse_bucket_uri": {
 				Type:     schema.TypeString,
 				Optional: true,
@@ -137,10 +157,6 @@ func DataflowInvokeRunResource() *schema.Resource {
 			},
 
 			// Computed
-			"archive_uri": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
 			"class_name": {
 				Type:     schema.TypeString,
 				Computed: true,
@@ -207,10 +223,6 @@ func DataflowInvokeRunResource() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-			"spark_version": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
 			"state": {
 				Type:     schema.TypeString,
 				Computed: true,
@@ -332,6 +344,11 @@ func (s *DataflowInvokeRunResourceCrud) Create() error {
 		request.ApplicationId = &tmp
 	}
 
+	if archiveUri, ok := s.D.GetOkExists("archive_uri"); ok {
+		tmp := archiveUri.(string)
+		request.ArchiveUri = &tmp
+	}
+
 	if arguments, ok := s.D.GetOkExists("arguments"); ok {
 		interfaces := arguments.([]interface{})
 		tmp := make([]string, len(interfaces))
@@ -372,6 +389,11 @@ func (s *DataflowInvokeRunResourceCrud) Create() error {
 		request.DriverShape = &tmp
 	}
 
+	if execute, ok := s.D.GetOkExists("execute"); ok {
+		tmp := execute.(string)
+		request.Execute = &tmp
+	}
+
 	if executorShape, ok := s.D.GetOkExists("executor_shape"); ok {
 		tmp := executorShape.(string)
 		request.ExecutorShape = &tmp
@@ -408,6 +430,11 @@ func (s *DataflowInvokeRunResourceCrud) Create() error {
 		}
 	}
 
+	if sparkVersion, ok := s.D.GetOkExists("spark_version"); ok {
+		tmp := sparkVersion.(string)
+		request.SparkVersion = &tmp
+	}
+
 	if warehouseBucketUri, ok := s.D.GetOkExists("warehouse_bucket_uri"); ok {
 		tmp := warehouseBucketUri.(string)
 		request.WarehouseBucketUri = &tmp
@@ -547,6 +574,10 @@ func (s *DataflowInvokeRunResourceCrud) SetData() error {
 		s.D.Set("driver_shape", *s.Res.DriverShape)
 	}
 
+	if s.Res.Execute != nil {
+		s.D.Set("execute", *s.Res.Execute)
+	}
+
 	if s.Res.ExecutorShape != nil {
 		s.D.Set("executor_shape", *s.Res.ExecutorShape)
 	}