title: Define KSQL STRUCT on JSON valued topic with different types
body: <p>(<strong>Edit</strong>: slight edits to better reflect intention, but large edit due to progress made.)</p>
  
  <p>A topic <code>"t_raw"</code> is given messages of multiple types, where they all contain a common <code>"type"</code> key:</p>
  
  <pre><code>{"type":"key1","data":{"ts":"2018-11-20 19:20:21.1","a":1,"b":"hello"}}
  {"type":"key2","data":{"ts":"2018-11-20 19:20:22.2","a":1,"c":11,"d":"goodbye"}}
  {"type":"key1","data":{"ts":"2018-11-20 19:20:23.3","a":2,"b":"hello2"}}
  {"type":"key2","data":{"ts":"2018-11-20 19:20:24.4","a":3,"c":22,"d":"goodbye2"}}
  </code></pre>
  
  <p>Ultimately, I need to split this into other streams where they will be chopped/aggregated/processed. I'd like to be able to use <code>STRUCT</code> for everything, but my current effort has me doing this:</p>
  
  <pre><code>create stream raw (type varchar, data varchar) \
    with (kafka_topic='t_raw', value_format='JSON');
  </code></pre>
  
  <p>for the first level, then</p>
  
  <pre><code>create stream key1 with (TIMESTAMP='ts', timestamp_format='yyyy-MM-dd HH:mm:ss.S') as \
    select \
      extractjsonfield(data, '$.ts') as ts, \
      extractjsonfield(data, '$.a') as a, extractjsonfield(data, '$.b') as b \
    from raw where type='key1';
  create stream key2 with (TIMESTAMP='ts', timestamp_format='yyyy-MM-dd HH:mm:ss.S') as \
    select \
      extractjsonfield(data, '$.ts') as ts, \
      extractjsonfield(data, '$.a') as a, extractjsonfield(data, '$.c') as c, \
      extractjsonfield(data, '$.d') as d \
    from raw where type='key2';
  </code></pre>
  
  <p>This seems to work, but with the recent addition of <code>STRUCT</code>, is there a way to use it in lieu of <code>extractjsonfield</code> as done above?</p>
  
  <pre><code>ksql&gt; select * from key1;
  1542741621100 | null | 2018-11-20 19:20:21.1 | 1 | hello
  1542741623300 | null | 2018-11-20 19:20:23.3 | 2 | hello2
  ^CQuery terminated
  ksql&gt; select * from key2;
  1542741622200 | null | 2018-11-20 19:20:22.2 | 1 | 11 | goodbye
  1542741624400 | null | 2018-11-20 19:20:24.4 | 3 | 22 | goodbye2
  </code></pre>
  
  <p>If not with <code>STRUCT</code>, is there a straight-forward way to do this with vanilla kafka-streams (vice <code>ksql</code>, ergo the <a href="/questions/tagged/apache-kafka-streams" class="post-tag" title="show questions tagged &#39;apache-kafka-streams&#39;" rel="tag">apache-kafka-streams</a> tag)?</p>
  
  <p>Is there a more kafka-esque/efficient/elegant way to parse this? 
  I cannot define it as an empty <code>STRUCT&lt;&gt;</code></p>
  
  <pre><code>ksql&gt; CREATE STREAM some_input ( type VARCHAR, data struct&lt;&gt; ) \
        WITH (KAFKA_TOPIC='t1', VALUE_FORMAT='JSON');
  line 1:52: extraneous input '&lt;&gt;' expecting {',', ')'}
  </code></pre>
  
  <p>There is <a href="https://github.com/confluentinc/ksql/issues/1538" rel="nofollow noreferrer">some (not-so-recent) discussion</a> on being able to do something like</p>
  
  <pre><code>CREATE STREAM key1 ( a INT, b VARCHAR ) AS \
    SELECT data-&gt;* from some_input where type = 'key1';
  </code></pre>
  
  <p>FYI: the above solution will not work in confluent-5.0.0, <a href="https://github.com/confluentinc/ksql/pull/1962" rel="nofollow noreferrer">a recent patch</a> fixed the <code>extractjsonfield</code> bug and enabled this solution.</p>
  
  <p>The real data has several more similar message types. They all contain <code>"type"</code> and <code>"data"</code> keys (and no others at the top-level), and almost all have the <code>"ts"</code> timestamp equivalent nested within <code>"data"</code>.</p>