If you have a stream that is either unkeyed (the key is null) or not keyed by the desired field, you can rekey the stream by issuing a CREATE STREAM AS SELECT (CSAS) statement and explicitly specifying the new key with PARTITION BY. The new stream can be partitioned by a value or a scalar function.
For example, suppose that you have an unkeyed stream representing movies:
CREATE STREAM movies (id INT, title VARCHAR, genre VARCHAR)
WITH (kafka_topic='movies', partitions=1, value_format = 'avro');
Assume that the title field includes both the title and release year, e.g.:
INSERT INTO movies (id, title, genre) VALUES (294, 'Die Hard::1988', 'action');
Then you can rekey by a value (e.g., the id field) as follows:
CREATE STREAM movies_by_id
WITH (KAFKA_TOPIC='movies_by_id') AS
SELECT *
FROM movies
PARTITION BY id;
Or, you can rekey by function (e.g, the title extracted from the title field) as follows:
CREATE STREAM movies_by_title
WITH (KAFKA_TOPIC='movies_by_title') AS
SELECT *
FROM movies
PARTITION BY SPLIT(title, '::')[1];