Flink CDC如何只取里面的INSERT和UPDATE操作，或者这么可以不回撤？

Flink CDC可以通过Flink SQL来处理Canal Json数据，只取其中的INSERT和UPDATE操作。具体来说，可以使用Flink SQL的DML语句来实现这个功能。

例如，可以使用以下语句来只获取INSERT和UPDATE操作：

SELECT * FROM canal_json WHERE operation = 'INSERT' OR operation = 'UPDATE';

此外，为了避免回撤时通过主键删除导致数据不完整的问题，可以考虑使用UPSERT（Update or Insert）操作来更新ES中的数据。这样，在更新数据时，如果数据已经存在，则更新它；否则，插入新数据。这可以避免因为先删除再插入而导致的数据丢失问题。

具体来说，可以使用Elasticsearch的API来实现UPSERT操作。例如，可以使用以下代码来更新或插入一条文档：

// 创建索引请求对象
IndexRequest indexRequest = new IndexRequest("index_name");
indexRequest.id("document_id");

// 创建文档对象
Map<String, Object> document = new HashMap<>();
document.put("field1", "value1");
document.put("field2", "value2");
// ...

// 将文档对象转换为JSON字符串
String jsonString = JSON.toJSONString(document);

// 创建更新请求对象
UpdateRequest updateRequest = new UpdateRequest("index_name", "document_id");
updateRequest.doc(jsonString, XContentType.JSON);

// 执行更新或插入操作
RestHighLevelClient client = new RestHighLevelClient(/*...*/);
client.index(updateRequest, RequestOptions.DEFAULT);

Flink CDC可以通过使用Flink SQL来过滤Canal Json中的INSERT和UPDATE操作，具体Flink CDC可以通过使用Flink SQL来过滤Canal Json中的INSERT和UPDATE操作，具体步骤如下：

1. 在Flink SQL中创建源表，将Canal Json数据作为源表的数据源。
2. 使用SELECT语句选择需要的数据，例如只选择INSERT和UPDATE操作。
3. 将结果插入到目标表中，例如ES。

为了避免回撤时通过主键删除导致数据不完整的问题，可以使用以下方法：

1. 在目标表中添加一个状态字段，用于记录每条数据的状态（例如未处理、已处理）。
2. 在INSERT和UPDATE操作中，根据主键更新状态字段的值。
3. 在查询数据时，只查询状态为“未处理”的数据。
4. 在处理完数据后，更新状态字段的值。

这样可以避免回撤时通过主键删除导致数据不完整的问题。