title: Kafka topic with Variable nested JSON object as KSQL DB stream
body: <p>I'm trying to join two existing Kafka topics in KSQL. Some data samples from Kafka (actual values redacted due to corporate environment):</p>
  <p>device topic:</p>
  <pre><code>{
    &quot;persistTime&quot; : &quot;2020-10-06T13:30:25.373Z&quot;,
    &quot;previous&quot; : {
      &quot;device&quot; : &quot;REDACTED&quot;,
      &quot;type&quot; : &quot;REDACTED&quot;,
      &quot;group&quot; : &quot;REDACTED&quot;,
      &quot;inventoryState&quot; : &quot;unknown&quot;,
      &quot;managementState&quot; : &quot;registered&quot;,
      &quot;communicationId&quot; : &quot;REDACTED&quot;,
      &quot;manufacturer&quot; : &quot;&quot;,
      &quot;description&quot; : &quot;&quot;,
      &quot;model&quot; : &quot;&quot;,
      &quot;location&quot; : {
        &quot;geo&quot; : {
          &quot;latitude&quot; : &quot;REDACTED&quot;,
          &quot;longitude&quot; : &quot;REDACTED&quot;
        },
        &quot;address&quot; : {
          &quot;city&quot; : &quot;&quot;,
          &quot;postalCode&quot; : &quot;&quot;,
          &quot;street&quot; : &quot;&quot;,
          &quot;houseNumber&quot; : &quot;&quot;,
          &quot;floor&quot; : &quot;&quot;,
          &quot;company&quot; : &quot;&quot;,
          &quot;country&quot; : &quot;&quot;,
          &quot;reference&quot; : &quot;&quot;,
          &quot;timeZone&quot; : &quot;&quot;,
          &quot;region&quot; : &quot;&quot;,
          &quot;district&quot; : &quot;&quot;
        },
        &quot;logicalInstallationPoint&quot; : &quot;&quot;
      },
      &quot;tags&quot; : [ ]
    },
    &quot;current&quot; : {
      &quot;device&quot; : &quot;REDACTED&quot;,
      &quot;type&quot; : &quot;REDACTED&quot;,
      &quot;group&quot; : &quot;REDACTED&quot;,
      &quot;inventoryState&quot; : &quot;unknown&quot;,
      &quot;managementState&quot; : &quot;registered&quot;,
      &quot;communicationId&quot; : &quot;REDACTED&quot;,
      &quot;manufacturer&quot; : &quot;&quot;,
      &quot;description&quot; : &quot;&quot;,
      &quot;model&quot; : &quot;&quot;,
      &quot;location&quot; : {
        &quot;geo&quot; : {
          &quot;latitude&quot; : &quot;REDACTED&quot;,
          &quot;longitude&quot; : &quot;REDACTED&quot;
        },
        &quot;address&quot; : {
          &quot;city&quot; : &quot;&quot;,
          &quot;postalCode&quot; : &quot;&quot;,
          &quot;street&quot; : &quot;&quot;,
          &quot;houseNumber&quot; : &quot;&quot;,
          &quot;floor&quot; : &quot;&quot;,
          &quot;company&quot; : &quot;&quot;,
          &quot;country&quot; : &quot;&quot;,
          &quot;reference&quot; : &quot;&quot;,
          &quot;timeZone&quot; : &quot;&quot;,
          &quot;region&quot; : &quot;&quot;,
          &quot;district&quot; : &quot;&quot;
        },
        &quot;logicalInstallationPoint&quot; : &quot;&quot;
      },
      &quot;tags&quot; : [ ]
    }
  }
  </code></pre>
  <p>device-event topic (1st sample):</p>
  <pre><code>{
    &quot;device&quot; : &quot;REDACTED&quot;,
    &quot;event&quot; : &quot;403151&quot;,
    &quot;firstOccurrenceTime&quot; : &quot;2020-09-30T11:03:50.000Z&quot;,
    &quot;lastOccurrenceTime&quot; : &quot;2020-09-30T11:03:50.000Z&quot;,
    &quot;occurrenceCount&quot; : 1,
    &quot;receiveTime&quot; : &quot;2020-09-30T11:03:50.000Z&quot;,
    &quot;persistTime&quot; : &quot;2020-09-30T14:32:59.580Z&quot;,
    &quot;state&quot; : &quot;open&quot;,
    &quot;context&quot; : {
      &quot;2&quot; : &quot;25&quot;,
      &quot;3&quot; : &quot;0&quot;,
      &quot;4&quot; : &quot;60&quot;,
      &quot;1&quot; : &quot;REDACTED&quot;
    }
  }
  </code></pre>
  <p>device-event topic (2nd sample):</p>
  <pre><code>{
    &quot;device&quot; : &quot;REDACTED&quot;,
    &quot;event&quot; : &quot;402004&quot;,
    &quot;firstOccurrenceTime&quot; : &quot;2020-10-07T07:02:48Z&quot;,
    &quot;lastOccurrenceTime&quot; : &quot;2020-10-07T07:02:48Z&quot;,
    &quot;occurrenceCount&quot; : 1,
    &quot;receiveTime&quot; : &quot;2020-10-07T07:02:48Z&quot;,
    &quot;persistTime&quot; : &quot;2020-10-07T07:15:28.533Z&quot;,
    &quot;state&quot; : &quot;open&quot;,
    &quot;context&quot; : {
      &quot;2&quot; : &quot;2020-10-07T07:02:48.0000000Z&quot;,
      &quot;1&quot; : &quot;REDACTED&quot;
    }
  }
  </code></pre>
  <p>The issue that I'm facing is the varying amount of variables inside of <code>context</code> under the device-event topic.</p>
  <p>I've tried the following statements for creating the events stream on ksqlDB:</p>
  <pre><code>CREATE STREAM &quot;events&quot;\
  (&quot;device&quot; VARCHAR, \
  &quot;event&quot; VARCHAR, \
  &quot;firstOccurenceTime&quot; VARCHAR, \
  &quot;lastOccurenceTime&quot; VARCHAR, \
  &quot;occurenceCount&quot; INTEGER, \
  &quot;receiveTime&quot; VARCHAR, \
  &quot;persistTime&quot; VARCHAR, \
  &quot;state&quot; VARCHAR, \
  &quot;context&quot; ARRAY&lt;STRING&gt;) \
  WITH (KAFKA_TOPIC='device-event', VALUE_FORMAT='JSON');
  </code></pre>
  <pre><code>CREATE STREAM &quot;events&quot;\
  (&quot;device&quot; VARCHAR, \
  &quot;event&quot; VARCHAR, \
  &quot;firstOccurenceTime&quot; VARCHAR, \
  &quot;lastOccurenceTime&quot; VARCHAR, \
  &quot;occurenceCount&quot; INTEGER, \
  &quot;receiveTime&quot; VARCHAR, \
  &quot;persistTime&quot; VARCHAR, \
  &quot;state&quot; VARCHAR, \
  &quot;context&quot; STRUCT\
  &lt;&quot;1&quot; VARCHAR, \
  &quot;2&quot; VARCHAR, \
  &quot;3&quot; VARCHAR, \
  &quot;4&quot; VARCHAR&gt;) \
  WITH (KAFKA_TOPIC='ext_device-event_10195', VALUE_FORMAT='JSON');
  </code></pre>
  <p>The second statement only brings in data that has all four context variables present (&quot;1&quot;, &quot;2&quot;, &quot;3&quot; and &quot;4&quot;).</p>
  <p>How would one go about creating the KSQL equivalent stream for the device-event Kafka topic?</p>