**Context**: 

We have a model which has 5 flows.

* Flow-1 : Input Data from different sources , these are input to Flow-2
* Flow-2, Flow-3 and Flow-4 : are ML models each dumping few fields to s3
* Flow-5 : Reporting Layer with data from output of Flow-2,Flow-3,Flow-4
* Overall data size is very small

**Problem**: 

* Flow-5 which is reporting layer based on few SQL with input data coming from Flow-2,Flow-3 and Flow-4.
* Flow-2,Flow-3 and Flow-4 has one common joining field and remaining fields are different.
* We can create a SQL joining Flow-2,3,4 data stored in three different tables with few calculations/Aggregation.However the number of output fields from Flow-2,3,4 may vary in each run.

	+ Problem-1: every time s3 file (Flow-2/3/4) structure changes, which often results in issue during COPY, as the target table schema differs from s3 file structire (to fix, need to manually add/delete fields in target table,to align with s3 data)
	+ Problem-2: for any additions/deletion in s3 files, need to make changes in reports by adding/removing column

**Approach**:

* SQL way - standardize s3 dump/target table/report SQL i.e. standardize the number of possible columns in each flow(2,3,4) output,also in the target table, so that if any fields are not available, just load them as NULL/blank during s3 dump and COPY as blank. Standardize target table structure aligned with s3 template. Also standardize reporting SQL
* SCALA/SPARK : Currently exploring this option. to perform a PoC, created two s3 dumps,created two dataframes in scala, tried dataframe joins as well as spark SQL join. Im still not sure if there is anyway we can dynamically pick new columns ie making the spark code generic.

	+ with creating dataframes directly pointing to s3, we can solve the COPY data (dynamic fields) to target table problem.
	+ however, the reporting SQL issue still persists (or atleast I dont know, need to find a way how can it be handled)

**Question**:

Is there anyway,can we handle the issue (dynamic column selection in SQL) in Scala/SparkSQL ?
Note: s3 file always contains only the required fields from each run.Issue is not reading the dynamic fields from s3 (which will be auto taken care by dataframe, instead issue is how can we make the SQL(saprkSQL/API) code to handle this)

**Example/Scenario-1**:

* **Run-1**

	+ df1= country,1st\_buy (here dataframe directly points to s3, which has only required attributes)
	+ df2= country,percent (here dataframe directly points to s3, which has only required attributes)

```
--Sample SQL code (need to convert to sparkSQL/dataframe API)
SELECT df1.1st_buy,
       df1.percent/df2.percent -- DERIVED FIELD
FROM df1,df2
WHERE df1.country=df2.country

```

* **Run-2** (here one additional column was added to df1)

	+ df1= country,1st\_buy,2nd\_buy (here dataframe directly points to s3, which has only required attributes)
	+ df2= country,percent (here dataframe directly points to s3, which has only required attributes)

```
****how can we handle this part of new field 2nd_buy dynamically****

--Sample SQL code (need to convert to sparkSQL/dataframe API)
SELECT df1.1st_buy,
       df1.2nd_buy,
       df1.1st_buy/df2.percent -- DERIVED FIELD
       df1.2nd_buy/df2.percent -- DERIVED FIELD
FROM df1,df2
WHERE df1.country=df2.country

```

**Example/Scenario-2**:

* **Run-1**

	+ df1= country,1st\_buy (here dataframe directly points to s3, which has only required attributes)

```
--Sample SQL
SELECT country,sum(df1.1st_buy)
FROM df1
GROUP BY country

--Dataframe API/SparkSQL
df1.groupBy("country").sum("1st_buy").show()

```

* **Run-2** (here one additional column was added to df1)

	+ df1= country,1st\_buy,2nd\_buy (here dataframe directly points to s3, which has only required attributes)

```
****how can we handle this part of new field 2nd_buy dynamically****

--Sample SQL
SELECT country,sum(df1.1st_buy),sum(2nd_buy)
FROM df1
GROUP BY country

--Dataframe API/SparkSQL
df1.groupBy("country").sum("1st_buy","2nd_buy").show() 

```