使用WHERE子句基于KSQL从Kafka Topic创建表时遇语法错误求助

你的语法错误根源是将「直接定义表结构绑定Kafka主题」和「通过查询流生成物化表」两种语法混写了，KSQL不支持这种合并写法，因此触发了as位置的语法匹配错误。

正确写法一：通过查询直接生成物化表（推荐） #

如果要从orders_inputs流筛选数据并生成orders表，同时指定底层Kafka主题的分区、副本数，使用以下语法：

CREATE TABLE orders 
WITH (
    KAFKA_TOPIC='ORDERS',
    PARTITIONS=1,
    REPLICAS=1,
    VALUE_FORMAT='JSON' -- 根据你的流数据格式调整，比如AVRO、DELIMITED等
) AS 
SELECT orderid, itemid 
FROM orders_inputs 
WHERE type='t1'
PARTITION BY orderid -- 若流的消息Key不是orderid，需用此指定表的主键对应字段
EMIT CHANGES;

• 无需手动定义表字段，KSQL会自动从SELECT语句推断字段类型和结构
• PARTITION BY用于指定表的主键（对应Kafka消息的Key），确保后续更新逻辑基于主键生效
• EMIT CHANGES表示持续从流中读取数据并更新表，是KSQL 0.29.2中持续查询的必填项

正确写法二：分两步创建表并插入数据 #

如果需要先手动定义表结构，再从流中导入数据，执行以下两步：

1. 先创建表结构并绑定Kafka主题：

CREATE TABLE orders (
    orderid varchar PRIMARY KEY,
    itemid varchar
) WITH (
    KAFKA_TOPIC='ORDERS',
    PARTITIONS=1,
    REPLICAS=1,
    VALUE_FORMAT='JSON'
);

2. 从流中筛选数据插入表：

INSERT INTO orders 
SELECT orderid, itemid 
FROM orders_inputs 
WHERE type='t1';

适配你的业务需求 #

针对你筛选eventType属性、存储客户数据的场景，只需调整SELECT语句中的字段和过滤条件即可，示例：

CREATE TABLE customer_info 
WITH (
    KAFKA_TOPIC='CUSTOMER_INFO',
    PARTITIONS=3,
    REPLICAS=2,
    VALUE_FORMAT='JSON'
) AS 
SELECT customer_id AS PRIMARY KEY, phone, email, address 
FROM customer_events_inputs 
WHERE eventType='CUSTOMER_UPDATE'
EMIT CHANGES;

内容的提问来源于stack exchange，提问作者Jose David Escobar A