Kafka consumer for debugging purposes: ``` docker run -it --rm \ confluentinc/cp-kafka \ kafka-console-consumer \ --bootstrap-server SSL://b-1.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,SSL://b-2.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,SSL://b-3.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094 \ --consumer-property security.protocol=SSL \ --consumer-property ssl.truststore.location=/usr/lib/jvm/zulu-8-amd64/jre/lib/security/cacerts \ --topic default_node_topic --from-beginning --max-messages 42 ``` Once you have Ksqldb server up and running, you'll see messages like these: ``` INFO 127.0.0.1 - - [27/Aug/2020:21:23:51 +0000] "POST /ksql HTTP/1.1" 200 1496 172 (io.confluent.rest-utils.requests:62) ``` Run Ksql cli in a separate container: ``` docker run -it confluentinc/cp-ksql-cli http://dev-httpd-myron:8088 ``` Check to see if your topics are visbile: ``` ksql> SHOW TOPICS; Kafka Topic | Partitions | Partition Replicas ----------------------------------------------------------------------- _kafka-connect-group-01-v04-configs | 1 | 3 _kafka-connect-group-01-v04-offsets | 25 | 3 _kafka-connect-group-01-v04-status | 5 | 3 _kafka-connect-group-02-v04-configs | 1 | 3 _kafka-connect-group-02-v04-offsets | 25 | 3 _kafka-connect-group-02-v04-status | 5 | 3 _kafka-connect-group-03-v04-configs | 1 | 1 _kafka-connect-group-03-v04-offsets | 25 | 1 _kafka-connect-group-03-v04-status | 5 | 1 _schemas | 1 | 3 default_node_topic | 1 | 3 default_relationship_topic | 1 | 3 maxwell_art_relations_artist_info | 3 | 3 maxwell_art_relations_multi_table | 3 | 3 maxwell_art_relations_neo4j_ddl | 3 | 3 maxwell_art_relations_neo4j_dlq | 3 | 3 orcd_test_1ksql_processing_log | 1 | 1 quickstart-config | 1 | 3 quickstart-offsets | 25 | 3 quickstart-status | 5 | 3 ----------------------------------------------------------------------- ``` Create a stream on one of your topics: ``` CREATE STREAM default_node_stream (payload STRUCT< after STRUCT< `properties` STRUCT< name VARCHAR, id BIGINT, vendorId BIGINT, subaccountId BIGINT, uuid VARCHAR>, labels ARRAY>, `type` VARCHAR> ) WITH (KAFKA_TOPIC = 'default_node_topic', VALUE_FORMAT = 'JSON'); ``` At this point you are ready to query your event data: ``` ksql> SELECT * >FROM default_node_stream >WHERE payload->after->`properties`->vendorId = 21989 >emit changes limit 5; +-------------------------------------+-------------------------------------+-------------------------------------+ |ROWTIME |ROWKEY |PAYLOAD | +-------------------------------------+-------------------------------------+-------------------------------------+ |1597886615832 |22731203-0 |{AFTER={properties={NAME=Emily Barker| | | |, ID=126893376, VENDORID=21989, SUBAC| | | |COUNTID=51571, UUID=48737884-3a4f-4f5| | | |c-ba64-ad24460cbd3a}, LABELS=[LabelPa| | | |rticipant, Orchard]}, type=node} | |1597886615961 |22731204-1 |{AFTER={properties={NAME=Greg Freeman| | | |, ID=126893377, VENDORID=21989, SUBAC| | | |COUNTID=51571, UUID=3967950e-409b-488| | | |a-be9f-cf9a92dc6fc6}, LABELS=[LabelPa| | | |rticipant, Orchard]}, type=node} | |1597886615962 |22731205-2 |{AFTER={properties={NAME=Jimmy Van He| | | |usen, ID=190653319, VENDORID=21989, S| | | |UBACCOUNTID=52562, UUID=643b58e5-7004| | | |-4f83-a612-42c5923a7f21}, LABELS=[Lab| | | |elParticipant, Orchard]}, type=node} | |1597886615962 |22731206-3 |{AFTER={properties={NAME=Aaron Einhou| | | |se, ID=191561484, VENDORID=21989, SUB| | | |ACCOUNTID=16209, UUID=aea22277-ba5d-4| | | |c84-91e4-e6d8c6a7c4f9}, LABELS=[Label| | | |Participant, Orchard]}, type=node} | |1597886615962 |22731207-4 |{AFTER={properties={NAME=Jonathan Sin| | | |gleton, ID=123662450, VENDORID=21989,| | | | SUBACCOUNTID=40789, UUID=f6abc9c1-89| | | |9e-4ab4-913f-8daed48969e4}, LABELS=[L| | | |abelParticipant, Orchard]}, type=node| | | |} | Limit Reached Query terminated ``` Here's an example of how you can transform the data: ``` ksql> SELECT payload->after->`properties`->uuid as UUID, > payload->after->`properties`->name as NAME, > payload->after->`properties`->vendorId as VENDORID, > payload->after->`properties`->subaccountId as SUBACCOUNTID >FROM default_node_stream >WHERE ARRAYCONTAINS(payload->after->labels, 'LabelParticipant') >emit changes limit 5; +---------------------------+---------------------------+---------------------------+---------------------------+ |UUID |NAME |VENDORID |SUBACCOUNTID | +---------------------------+---------------------------+---------------------------+---------------------------+ |48737884-3a4f-4f5c-ba64-ad2|Emily Barker |21989 |51571 | |4460cbd3a | | | | |3967950e-409b-488a-be9f-cf9|Greg Freeman |21989 |51571 | |a92dc6fc6 | | | | |643b58e5-7004-4f83-a612-42c|Jimmy Van Heusen |21989 |52562 | |5923a7f21 | | | | |aea22277-ba5d-4c84-91e4-e6d|Aaron Einhouse |21989 |16209 | |8c6a7c4f9 | | | | |f6abc9c1-899e-4ab4-913f-8da|Jonathan Singleton |21989 |40789 | |ed48969e4 | | | | Limit Reached Query terminated ``` At this point you are ready to create your new stream that can be used to send data over to something like Elasticsearch where document keys are important to perform upserts. ``` CREATE STREAM label_participant_uuid_test_rekeyed AS SELECT payload->after->`properties`->uuid as UUID, payload->after->`properties`->name as NAME, payload->after->`properties`->vendorId as VENDORID, payload->after->`properties`->subaccountId as SUBACCOUNTID FROM default_node_stream WHERE ARRAYCONTAINS(payload->after->labels, 'LabelParticipant') AND payload->after->`properties`->uuid IS NOT NULL PARTITION BY UUID; ``` Check to make sure the Kafka topic key has now been modified: ``` ksql> SELECT * >FROM label_participant_uuid_test_rekeyed >emit changes limit 5; +-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+ |ROWTIME |ROWKEY |UUID |NAME |VENDORID |SUBACCOUNTID | +-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+ |1599046584955 |5d1621ea-1fb5-472|5d1621ea-1fb5-472|DUA LIPA, |21989 |44281 | | |f-bc6d-0f3c8363d2|f-bc6d-0f3c8363d2| | | | | |f3 |f3 | | | | |1599046621118 |63a026ca-ccbe-4f7|63a026ca-ccbe-4f7|Anton Delost |21989 |53493 | | |0-bece-bd00a0fbdf|0-bece-bd00a0fbdf| | | | | |ff |ff | | | | |1599046621118 |3cb256bd-1e09-429|3cb256bd-1e09-429|Will Oldham |21989 |52979 | | |7-a3be-929d0c4833|7-a3be-929d0c4833| | | | | |9b |9b | | | | |1599046621119 |2a3878cb-d5c9-4d6|2a3878cb-d5c9-4d6|TDK |21989 |52422 | | |5-8924-6d048c6695|5-8924-6d048c6695| | | | | |7b |7b | | | | |1599046621119 |08114b27-4fee-482|08114b27-4fee-482|Zac Rae |21989 |52979 | | |2-9a16-7e33f597d8|2-9a16-7e33f597d8| | | | | |97 |97 | | | | ``` More instructions about how to setup the ES sink connector can be found here: https://github.com/theorchard/collab/tree/master/mymac80/kafka-connect-elasticsearch