ksql_extra_twitter_filter.sql

CREATE STREAM twitter_reinvent_kafka_ksql AS 
SELECT 
  TIMESTAMPTOSTRING(CreatedAt, 'yyyy-MM-dd HH:mm:ss.SSS') AS CreatedAt, 
  EXTRACTJSONFIELD(user,'$.Name') AS user_Name, 
  EXTRACTJSONFIELD(user,'$.ScreenName') AS user_ScreenName, 
  EXTRACTJSONFIELD(user,'$.Location') AS user_Location, 
  EXTRACTJSONFIELD(user,'$.Description') AS user_Description, 
  Text,
  hashtagentities,
  lang 
FROM twitter_raw 
WHERE LCASE(hashtagentities) LIKE '%kafkasummit%' AND 
     (LCASE(hashtagentities) LIKE '%ksql%' OR 
      LCASE(hashtagentities) LIKE '%apachekafka%');