使用elasticsearch-java向Elasticsearch数据流批量插入报错求助

问题原因 #

报错only write ops with an op type of create are allowed in data-streams是因为Elasticsearch数据流的设计仅支持create类型的写入操作，不允许默认的index操作（该操作会在文档存在时更新、不存在时创建）。你当前代码使用的op.index(...)默认采用index操作类型，因此触发了限制。

修改方案 #

只需将批量操作的写入类型改为create即可，有两种实现方式：

方式1：直接使用create操作 #

替换原代码中的.index(...)为.create(...)，示例代码如下：

List<Product> products = fetchProducts();

BulkRequest.Builder br = new BulkRequest.Builder();

for (Product product : products) {
    br.operations(op -> op           
        .create(idx -> idx            
            .index("products")       
            .id(product.getSku())
            .document(product)
        )
    );
}

BulkResponse result = esClient.bulk(br.build());

// 错误日志处理
if (result.errors()) {
    logger.error("批量插入出现错误");
    for (BulkResponseItem item : result.items()) {
        if (item.error() != null) {
            logger.error(item.error().reason());
        }
    }
}

方式2：在index操作中显式指定opType #

如果希望保留index的写法，可以在IndexRequest构建器中添加.opType(OpType.CREATE)指定操作类型，代码如下：

List<Product> products = fetchProducts();

BulkRequest.Builder br = new BulkRequest.Builder();

for (Product product : products) {
    br.operations(op -> op           
        .index(idx -> idx            
            .index("products")       
            .id(product.getSku())
            .document(product)
            .opType(OpType.CREATE) // 显式设置操作类型为CREATE
        )
    );
}

BulkResponse result = esClient.bulk(br.build());

// 错误日志处理
if (result.errors()) {
    logger.error("批量插入出现错误");
    for (BulkResponseItem item : result.items()) {
        if (item.error() != null) {
            logger.error(item.error().reason());
        }
    }
}

注意事项 #

• 数据流以追加写入时序数据为设计目标，create操作会在文档ID已存在时返回错误，这符合数据流的使用规范。
• 确保Product类能被Elasticsearch Java客户端正确序列化，若使用默认Jackson序列化，需保证类字段有合理的访问权限（如提供getter方法）或添加对应注解。

内容的提问来源于stack exchange，提问作者Ssk