title: Fanout data from Kafka to S3 using kafka connect
body: <p>My kafka gets a wrapper payload in Json format. The wrapper payload looks like this:</p>
  <pre><code>{
  &quot;format&quot;: &quot;wrapper&quot;,
  &quot;time&quot;: 1626814608000,
  &quot;events&quot;: [
    {
      &quot;id&quot;: &quot;item1&quot;,
      &quot;type&quot;: &quot;product1&quot;,
      &quot;count&quot;: 200
    },
    {
      &quot;id&quot;: &quot;item2&quot;,
      &quot;type&quot;: &quot;product2&quot;,
      &quot;count&quot;: 300
    }
  ],
  &quot;metadata&quot;: {
    &quot;schema&quot;: &quot;schema-1&quot;
  }
  </code></pre>
  <p>}</p>
  <p>I should export this to S3. But the catch here is, I should not store the wrapper. Instead, I should be storing the individual events based on the item.
  For example, it should be stored in S3 as follows:</p>
  <p><code>bucket/product1</code>:</p>
  <pre><code>{&quot;id&quot;: &quot;item1&quot;, &quot;type&quot;: &quot;product1&quot;, &quot;count&quot;: 200}
  </code></pre>
  <p><code>bucket/product2</code>:</p>
  <pre><code>{&quot;id&quot;: &quot;item2&quot;, &quot;type&quot;: &quot;product2&quot;, &quot;count&quot;: 300}
  </code></pre>
  <p>If you notice, the input is the wrapper with those events internally. However, my output should be each of those individual events stored in S3 in the same bucket with the <code>product type</code> as prefix.</p>
  <p>My question is, is it possible to use Kafka Connect to do this? I see they have this <a href="https://docs.confluent.io/platform/current/connect/transforms/overview.html" rel="nofollow noreferrer">Single Message transformer</a> which seems to be a way to mutate data inside the object, but not to fanout like what I want. Even the signature looks like an <code>R=&gt;R</code>
  <a href="https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/transforms/Transformation.java" rel="nofollow noreferrer">https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/transforms/Transformation.java</a></p>
  <p>So based on my research, it does not seem possible. But I want to check if I am missing something before using a different option.</p>