KsqlDB.sql

CREATE STREAM "my-stream1"
      (data STRUCT<
            EmailTemplateName VARCHAR, 
            EmailAddress VARCHAR>, 
       maskEmail VARCHAR, 
       market VARCHAR) 
  WITH (kafka_topic='growthnonprod-signup-api-emails-dev', value_format='json');

select data->EmailTemplateName as EmailTemplateName, data->EmailAddress as EmailAddress, maskEmail, market from "my-stream1" EMIT CHANGES;