如何从 ksqldb 表中删除一个值或插入一个墓碑值？答案

【问题标题】：How to delete a value from ksqldb table or insert a tombstone value?如何从 ksqldb 表中删除一个值或插入一个墓碑值？
【发布时间】：2021-05-24 02:16:36
【问题描述】：

如何通过 Rest api 或至少作为 ksqldb-cli 中的语句将 ksql 表中的行标记为删除？

CREATE TABLE movies (
      title VARCHAR PRIMARY KEY,
      id INT,
      release_year INT
    ) WITH (
      KAFKA_TOPIC='movies',
      PARTITIONS=1,
      VALUE_FORMAT = 'JSON'
    );

INSERT INTO MOVIES (ID, TITLE, RELEASE_YEAR) VALUES (48, 'Aliens', 1986);

由于显而易见的原因，这不起作用，但是 ksqldb 中不存在 DELETE 语句：

INSERT INTO MOVIES (ID, TITLE, RELEASE_YEAR) VALUES (48, null, null);

有没有办法创建推荐的 tombstone null 值，还是我需要将其直接写入底层主题？

【问题讨论】：

标签： ksqldb

【解决方案1】：

有一种方法可以解决这个问题。诀窍是使用KAFKA 值格式将墓碑写入底层主题。

这是一个使用原始 DDL 的示例。

-- Insert a second row of data
INSERT INTO MOVIES (ID, TITLE, RELEASE_YEAR) VALUES (42, 'Life of Brian', 1986);

-- Query table
ksql> SET 'auto.offset.reset' = 'earliest';

ksql> select * from movies emit changes limit 2;
+--------------------------------+--------------------------------+--------------------------------+
|TITLE                           |ID                              |RELEASE_YEAR                    |
+--------------------------------+--------------------------------+--------------------------------+
|Life of Brian                   |42                              |1986                            |
|Aliens                          |48                              |1986                            |
Limit Reached
Query terminated

现在声明一个新的流，它将使用相同的键写入相同的 Kafka 主题：

CREATE STREAM MOVIES_DELETED (title VARCHAR KEY, DUMMY VARCHAR) 
  WITH (KAFKA_TOPIC='movies', 
       VALUE_FORMAT='KAFKA');

插入墓碑消息：

INSERT INTO MOVIES_DELETED (TITLE,DUMMY) VALUES ('Aliens',CAST(NULL AS VARCHAR));

再次查询表：

ksql> select * from movies emit changes limit 2;
+--------------------------------+--------------------------------+--------------------------------+
|TITLE                           |ID                              |RELEASE_YEAR                    |
+--------------------------------+--------------------------------+--------------------------------+
|Life of Brian                   |42                              |1986                            |

检查基本主题

ksql> print movies;
Key format: KAFKA_STRING
Value format: JSON or KAFKA_STRING
rowtime: 2021/02/22 11:01:05.966 Z, key: Aliens, value: {"ID":48,"RELEASE_YEAR":1986}, partition: 0
rowtime: 2021/02/22 11:02:00.194 Z, key: Life of Brian, value: {"ID":42,"RELEASE_YEAR":1986}, partition: 0
rowtime: 2021/02/22 11:04:52.569 Z, key: Aliens, value: <null>, partition: 0

【讨论】：

感谢 Robin 提供的解决方法。这应该报告给 ksqldb 团队。插入删除的墓碑值应该更容易（没有中间流或值格式）。我没有以这种方式在push query中收到响应json对象中的row.tombstone。
如果您认为正确，请随时投票/标记为正确；) 您可以在此处记录问题并建议对 ksqlDB 的改进：github.com/confluentinc/ksql
当然我支持你，我只是想知道为什么我没有得到类似于此的行 {"row":{"columns":['Alien', null,null],"tombstone ":true}} 通过 Rest API，因为这是我最初的问题 :) 下次我会更冗长，但我希望它会在我实现插入墓碑值之后起作用。再次感谢
我使用的是 KSQL 6.2.0，我的表有复合键。我可以在 PRINT {tbl} FROM BEGINNING 中看到以下内容；行时间：2021/08/08 17:43:27.579 Z，键：{"r_n":"123-234","k_n":1,"k_2":1,"country":"IND","t" :"image","d":"2021-08-08"}, value: , partition: 0 但是表中仍然存在数据。这里有什么帮助吗？
+ 甚至其余的 api 都在返回没有 tombstone 属性的数据。