让我假设您的Kafka主题的消息（json格式）如下：

{
    "customer_id":value,
    "customer_name":value,
    "purchase_id":value,
    "purchase_desc":value,
    "product_id":value,
    "product_name":value
}

假设主题名为 my_topic .
所以你可以这样做：
为了 customer 流：

CREATE STREAM customer (
    -- which 'customer_id' match the JSON key
    customer_id BIGINT,
    customer_name VARCHAR
) WITH (
    VALUE_FORMAT = 'JSON',
    KAFKA_TOPIC = 'my_topic'
);

为了 purchase 表格：

CREATE STREAM purchase (
    purchase_id BIGINT,
    purchase_desc VARCHAR
) WITH (
    VALUE_FORMAT = 'JSON',
    KAFKA_TOPIC = 'my_topic'
);

为了 product 表格：

CREATE STREAM product (
    product_id BIGINT,
    product_name VARCHAR
) WITH (
    VALUE_FORMAT = 'JSON',
    KAFKA_TOPIC = 'my_topic'
);