Elasticsearch对多个文档进行索引的简便方法。Bulk api的支持可以实现一次请求执行批量的添加、删除、更新等操作。Bulk操作使用的是UDP协议,UDP无法确保与ElasticSearch服务器通信时不丢失数据。
将数据累计到一定数量进行批量上传,大大加快数据上传速度。
下面代码中,在 “具体操作添加到批量请求中” 前可以是新增、删除、更新等操作。
public class Bulk implements WriteRequest{
public static void main(String[] args) {
String globalname = "";
String valuep = "";
String time = "";
//数据条数记步
int count = 0;
//数据上传条数设定
int BULK_SIZE = 100;
try{
/*创建客户端*/
//client startup
//设置集群名称
Settings settings = Settings.builder()
.put("cluster.name", "elsearch")
.put("client.transport.sniff", true)
.build();
//创建client
TransportClient client = new
PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("10.122.4.79"),9300));
for(int i = 0; i<10; i++){
BulkRequestBuilder bulkRequest = client.prepareBulk();
bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
XContentBuilder builder=XContentFactory.jsonBuilder()
.startObject();
//TODO: 设置属性和值
builder.field("pointid",globalname);
builder.field("pointvalue",valuep);
builder.field("inputtime",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(time));
builder.endObject();
// 添加索引数据
IndexRequestBuilder requestBuilder = client
.prepareIndex("pointdata", "pointdata")
.setSource(builder);
// 具体操作添加到批量请求中
bulkRequest.add(requestBuilder);
count++;
if(count % BULK_SIZE == 0){
BulkResponse bulkResponse = bulkRequest
.execute().actionGet();
if (bulkResponse.hasFailures()) {
// LOGGER.error("导入索引数据失败: "+"pointdata");
// LOGGER.error("导入索引数据失败: "+bulkResponse.buildFailureMessage());
System.out.println("导入索引数据失败");
}
bulkRequest = client.prepareBulk();
bulkRequest.setRefreshPolicy(RefreshPolicy.IMMEDIATE);
count = 0;
}
}
}catch(Exception e){
e.printStackTrace();
}
}
@Override
public void readFrom(StreamInput arg0) throws IOException {
// TODO Auto-generated method stub
}
@Override
public void writeTo(StreamOutput arg0) throws IOException {
// TODO Auto-generated method stub
}
@Override
public RefreshPolicy getRefreshPolicy() {
// TODO Auto-generated method stub
return null;
}
@Override
public WriteRequest setRefreshPolicy(RefreshPolicy arg0) {
// TODO Auto-generated method stub
return null;
}
@Override
public ActionRequestValidationException validate() {
// TODO Auto-generated method stub
return null;
}
}
今天的文章Elasticsearch Java API 的使用(9)— Bluk的批量原理 & Bulk大数据量的批量上传示例「建议收藏」分享到此就结束了,感谢您的阅读。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/88770.html