title: Spark (Scala) conditionally write JSON to multiple dynamic output locations
body: <p>I have a problem to solve where data comes in as JSON from kinesis like below:</p>
  <pre><code>{
   datatype: &quot;datatype_1&quot;
   id : &quot;id_1&quot;
   data : {...}
  }
  </code></pre>
  <p>All records in the stream then need to go through a lookup function with datatype and id passed as arguments to find a unique group of locations to write the items to in JSON.</p>
  <p>i.e.</p>
  <pre><code>def get_locations(id: String, datatype: String): Array[String] = //custom logic here
  </code></pre>
  <p>where the resultant array would look like</p>
  <pre><code> [ &quot;s3:///example_bucket/example_folder_1&quot;, &quot;s3:///example_bucket2/example_folder_2&quot;]
  </code></pre>
  <p>My question is how do I most efficiently group records coming off the stream by datatype and id and write to the various s3 locations. I was hoping to do something like below:</p>
  <pre><code>        sparkSession.readStream.format(&quot;kinesis&quot;)
            .option(&quot;streamName&quot;, kinesis_stream_name)
            .option(&quot;initialPosition&quot;, &quot;latest&quot;)
            .option(&quot;region&quot;, aws_region)
            .load()
            //more transforms
            .select(
              col(&quot;datatype&quot;),
              col(&quot;id&quot;),
              col(&quot;data&quot;)
            )
  
  // Not sure how I can do what's below
  
  //          .write.partitionBy(&quot;id&quot;, &quot;datatype&quot;)
  //          .format(&quot;json&quot;)
  //          .option(&quot;compression&quot;,&quot;gzip&quot;)
  //          .save(get_locations(&quot;id&quot;,&quot;datatype&quot;)) //saving to all locations in result array
  </code></pre>