【问题标题】:How can there be duplicate IDs in a KSQL table?KSQL 表中怎么会有重复的 ID?
【发布时间】:2020-05-25 07:58:53
【问题描述】:

我正在使用 Confluent Community 和 Postgres 数据库,遇到以下问题。

事件很好地流入 kafka 并创建了主题。我从一个主题创建了一个流并重新设置了密钥,因为密钥为空。

根据重新加密流的新主题,我创建了一个表格。目标是拥有一个不断更新的对象表(此处为类别)。

问题是当我在数据库中执行手动更新时,表永远不会更新为新数据。行只是不断添加,就像它是一个流。当然,我再次进行了选择,因为我知道当我们仍在运行查询时会显示“更新”行。

ksql> select * from categories;
1568287458487 | 1 | 1 | Beverages | Soft drinks, coffees, teas, beers, and ales
1568287458487 | 2 | 2 | Condiments | Sweet and savory sauces, relishes, spreads, and seasonings
1568287458488 | 3 | 3 | Confections | Desserts, candies, and sweet breads
1568287458488 | 4 | 4 | Dairy Products | Cheeses
1568287458488 | 5 | 5 | Grains/Cereals | Breads, crackers, pasta, and cereal
1568287458488 | 6 | 6 | Meat/Poultry | Prepared meats
1568287458489 | 7 | 7 | Produce | Dried fruit and bean curd
1568287458489 | 8 | 8 | Seafood | Seaweed and fish
1568288647248 | 8 | 8 | Seafood2 | Seaweed and fish
1568290562250 | 1 | 1 | asdf | Soft drinks, coffees, teas, beers, and ales
1568296165250 | 8 | 8 | Seafood3 | Seaweed and fish
1568296704747 | 8 | 8 | Seafood4 | Seaweed and fish
^CQuery terminated
ksql> select * from categories;
1568287458487 | 1 | 1 | Beverages | Soft drinks, coffees, teas, beers, and ales
1568287458487 | 2 | 2 | Condiments | Sweet and savory sauces, relishes, spreads, and seasonings
1568287458488 | 3 | 3 | Confections | Desserts, candies, and sweet breads
1568287458488 | 4 | 4 | Dairy Products | Cheeses
1568287458488 | 5 | 5 | Grains/Cereals | Breads, crackers, pasta, and cereal
1568287458488 | 6 | 6 | Meat/Poultry | Prepared meats
1568287458489 | 7 | 7 | Produce | Dried fruit and bean curd
1568287458489 | 8 | 8 | Seafood | Seaweed and fish
1568288647248 | 8 | 8 | Seafood2 | Seaweed and fish
1568290562250 | 1 | 1 | asdf | Soft drinks, coffees, teas, beers, and ales
1568296165250 | 8 | 8 | Seafood3 | Seaweed and fish
1568296704747 | 8 | 8 | Seafood4 | Seaweed and fish
^CQuery terminated
ksql> 

postgres 中的类别表:

CREATE TABLE categories (
    category_id smallint NOT NULL,
    category_name character varying(15) NOT NULL,
    description text
);

KSQL 中的类别表:

ksql> describe extended categories;

Name                 : CATEGORIES
Type                 : TABLE
Key field            : CATEGORY_ID_ST
Key format           : STRING
Timestamp field      : Not set - using <ROWTIME>
Value format         : AVRO
Kafka topic          : categories_rk (partitions: 1, replication: 1)

 Field          | Type                      
--------------------------------------------
 ROWTIME        | BIGINT           (system) 
 ROWKEY         | VARCHAR(STRING)  (system) 
 CATEGORY_ID_ST | VARCHAR(STRING)           
 CATEGORY_NAME  | VARCHAR(STRING)           
 DESCRIPTION    | VARCHAR(STRING)           
 MESSAGETOPIC   | VARCHAR(STRING)           
 MESSAGESOURCE  | VARCHAR(STRING)           
--------------------------------------------

本应具有唯一 ROWKEY 的表怎么可能不断添加更多具有相同 ROWKEY 的“更新”行?

我实际上希望该表显示始终最新的类别列表,如https://www.youtube.com/watch?v=DPGn-j7yD68&list=PLa7VYi0yPIH2eX8q3mPpZAn3qCS1eDX8W&index=9 所述:

“TABLE 是事件的物化视图,每个键只有最新值”。但也许我误解了?

【问题讨论】:

  • 欢迎来到 StackOverflow!您能否分享您正在运行的 DDL 以及您如何将数据从 postgres 获取到 Kafka 的详细信息?
  • Postgres 通过 debezium 流入 kafka。使用 DDL 更新问题。
  • 请您也分享一下您的 KSQL DDL
  • 感谢您的帮助罗宾。我用其他数据更新了问题。

标签: apache-kafka confluent-platform ksqldb


【解决方案1】:

随着新数据的到来,KSQL 中的表会不断更新。表行写入的输出主题称为更改日志:它是表更改的不可变日志。如果特定键被多次更新,则输出主题将包含同一键的多条消息。每个新值都会替换最后一个。

当您运行如下查询时:

select * from categories;

在您使用的 ksql 版本中,您不会像在传统 RDBS 中所期望的那样运行传统查询。这样的查询将为您提供表中的当前行集。在 ksql 中,上述查询将在发生时将所有更新流式传输到行。因此,如果同一个密钥被多次更新;您将多次看到查询输出相同的键。

在更新版本的 ksqlDB 中,不会编写上述查询:

select * from categories emit changes;

在 ksql 内部,每个键只存储在物化表中一次,并且始终是看到的最新版本。

【讨论】:

    猜你喜欢
    • 2020-06-24
    • 2017-07-27
    • 1970-01-01
    • 2012-07-06
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2015-08-18
    相关资源
    最近更新 更多