Как указывает @ cricket_007, существует открытый тикет для обработки значений NULL таким образом в KSQL.
Один из способов, который вы можете использовать, - использовать INSERT INTO.Это не очень элегантно и, конечно, не так гибко, как такая функция, как COALESCE
:
# Set up some sample data, run this from bash
# For more info about kafkacat see
# https://docs.confluent.io/current/app-development/kafkacat-usage.html
kafkacat -b kafka-broker:9092 \
-t topic_with_nulls \
-P <<EOF
{"col1":1,"col2":16000,"col3":"foo"}
{"col1":2,"col2":42000}
{"col1":3,"col2":94000,"col3":"bar"}
{"col1":4,"col2":12345}
EOF
Вот обходной путь KSQL для обработки значений NULL в col3
:
-- Register the topic
CREATE STREAM topic_with_nulls (COL1 INT, COL2 INT, COL3 VARCHAR) \
WITH (KAFKA_TOPIC='topic_with_nulls',VALUE_FORMAT='JSON');
-- Query the topic to show there are some null values
ksql> SET 'auto.offset.reset'='earliest';
Successfully changed local property 'auto.offset.reset' from 'null' to 'earliest'
ksql> SELECT COL1, COL2, COL3 FROM topic_with_nulls;
1 | 16000 | foo
2 | 42000 | null
3 | 94000 | bar
4 | 12345 | null
-- Create a derived stream, with just records with no NULLs in COL3
CREATE STREAM NULL_WORKAROUND AS \
SELECT COL1, COL2, COL3 FROM topic_with_nulls WHERE COL3 IS NOT NULL;
-- Insert into the derived stream any records where COL3 *is* NULL, replacing it with a fixed string
INSERT INTO NULL_WORKAROUND \
SELECT COL1, COL2, 'N/A' AS COL3 FROM topic_with_nulls WHERE COL3 IS NULL;
-- Confirm that the NULL substitution worked
ksql> SELECT COL1, COL2, COL3 FROM NULL_WORKAROUND;
1 | 16000 | foo
2 | 42000 | N/A
3 | 94000 | bar
4 | 12345 | N/A