It’s no secret that automation is essential to streamlined development. Nowadays, we typically collaborate across teams and time zones, which requires a structured deployment process to stay organized. <\/p>\n\n\n\n
That’s why we decided to build a proof-of-concept using KSQL for data transformation and document our journey. While exploring the abilities of Confluent technologies was exciting, developing a build automation pipeline proved to be a challenge. This post will outline our solution for managing the deployment of KSQL streams and tables. <\/p>\n\n\n\n
To integrate with our existing CI\/CD, we used Jenkins pipelines to run Ansible playbooks, but these concepts can be applied to other configuration tools. <\/p>\n\n\n\n
Our project manipulated data from several input topics, creating a sequence of streams and tables to calculate results. Each query was built on previous ones, so if a change was made to the KSQL syntax (e.g. modifying a time window, or changing the type of join), then the entire sequence had to be torn down and rebuilt from scratch. <\/p>\n\n\n\n
To illustrate this concept, consider a simple example with just two queries. The first creates a stream from an existing input topic, and the second re-partitions the stream: <\/p>\n\n\n\n
CREATE STREAM PERSON_IDENTIFIER (first VARCHAR, last VARCHAR) \n WITH (KAFKA_TOPIC='person_topic', VALUE_FORMAT='JSON', KEY='last'); \nCREATE STREAM PERSON_IDENTIFIER_FIRST \n AS SELECT * FROM PERSON_IDENTIFIER PARTITION BY first; <\/code><\/pre><\/div>\n\n\n\nThese queries can be placed in a .sql file to run using KSQL’s RUN SCRIPT command. <\/p>\n\n\n\n
Tearing down looks similar, except the streams must be dropped in the opposite order, because the second depends on the first: <\/p>\n\n\n\n
DROP STREAM IF EXISTS PERSON_IDENTIFIER_FIRST DELETE TOPIC; \nDROP STREAM IF EXISTS PERSON_IDENTIFIER; <\/code><\/pre><\/div>\n\n\n\nIn an Ansible playbook, copy the setup and teardown scripts to the host, then run them using RUN SCRIPT. <\/p>\n\n\n\n
- name: "Tear down and set up KSQL queries" \n hosts: "{{ hostname }}" \n any_errors_fatal: True \n\n tasks: \n - name: "Copy teardown script file to host" \n copy: \n src: \/dir\/ksql-teardown-commands.sql \n dest: "\/dir\/ksql-teardown-commands.sql" \n\n - name: "Run teardown commands in KSQL" \n shell: ".\/current\/bin\/ksql http:\/\/{{ hostname }}:8088 <<< \\"RUN SCRIPT '\/dir\/ksql-teardown-commands.sql';\\"" \n\n - name: "Copy setup script file to host" \n copy: \n src: \/dir\/ksql-setup-commands.sql \n dest: "\/dir\/ksql-setup-commands.sql" \n\n - name: "Run setup commands in KSQL" \n shell: ".\/current\/bin\/ksql http:\/\/{{ hostname }}:8088 <<< \\"RUN SCRIPT '\/dir\/ksql-setup-commands.sql';\\"" <\/code><\/pre><\/div>\n\n\n\nThis playbook can be run from the command line or triggered using an automation server such as Jenkins. <\/p>\n\n\n\n
Terminating the queries<\/strong> <\/h2>\n\n\n\nThe above commands handle the destruction and reconstruction of streams, but they fail to terminate the underlying queries. <\/p>\n\n\n\n
Attempting to drop PERSON_IDENTIFIER will result in an error because the CSAS query used to construct PERSON_IDENTIFIER_FIRST is still running. <\/p>\n\n\n\n
As of Confluent 5.3, it is not possible to automatically terminate a query when a stream or table is dropped, nor is bulk termination supported. The simplest automated solution is to fetch the queries and terminate them one by one: <\/p>\n\n\n\n
- name: "Terminate running queries on all hosts in group" \n hosts: "{{ hostname }}" \n any_errors_fatal: True \n\n tasks: \n # Fetch and terminate currently running queries \n - name: "Get currently running queries via REST" \n uri: \n url: "http:\/\/{{ hostname }}:8088\/ksql" \n method: POST \n body: "{\\"ksql\\": \\"SHOW QUERIES;\\"}" \n headers: \n Content-Type: "application\/vnd.ksql.v1+json; charset=utf-8" \n return_content: yes \n body_format: json \n register: all_queries \n\n - name: "Parse query ids" \n set_fact: \n query_ids: "{{ all_queries | to_json | from_json | json_query(query) | list }}" \n vars: \n query: "json[0].queries[?starts_with(id,'CSAS_PERSON_') || starts_with(id,'CTAS_PERSON_')].id" \n delegate_to: 127.0.0.1 \n\n - name: "Terminate queries by id" \n shell: ".\/current\/bin\/ksql http:\/\/{{ hostname }}:8088 <<< \\"TERMINATE {{ item }};\\"" \n with_items: "{{ query_ids }}" <\/code><\/pre><\/div>\n\n\n\nWe use REST once again, this time fetching all of the currently running queries, then filtering for the ones associated with this project (i.e. beginning with CSAS_PERSON_ or CTAS_PERSON_). Then, the TERMINATE command is issued for each query. <\/p>\n\n\n\n
If a query fails to terminate, the error will not be detected by Ansible since the command was run in shell. As we only want to proceed if all relevant queries are terminated, we repeat the process of fetching and filtering, triggering the playbook to fail if the filtered list is not empty: <\/p>\n\n\n\n