官方地址:
https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs.html

通过JAVA api实现对ES的增删改查.

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
public class Test {
public static void main(String[] args)
throws ClassNotFoundException, SQLException, IOException, InterruptedException {
// 准备数据
long start = System.currentTimeMillis();
List<Map<String, ?>> data = prepareData();
long end = System.currentTimeMillis();
System.out.printf("准备数据用时:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
index(data);
end = System.currentTimeMillis();
System.out.printf("单条索引:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
update(data);
end = System.currentTimeMillis();
System.out.printf("单条更新:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
delete(data);
end = System.currentTimeMillis();
System.out.printf("单条删除:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
bulk(data);
end = System.currentTimeMillis();
System.out.printf("批量索引:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
get(data);
end = System.currentTimeMillis();
System.out.printf("ID查询:%s秒\n", (end - start) / 1000);
start = System.currentTimeMillis();
bulkProcessor(data);
end = System.currentTimeMillis();
System.out.printf("bulkProcessor处理:%s秒\n", (end - start) / 1000);
}
private static List<Map<String, ?>> prepareData()
throws ClassNotFoundException, SQLException, UnknownHostException, InterruptedException {
List<Map<String, ?>> data = new ArrayList<Map<String, ?>>();
Class.forName("com.mysql.jdbc.Driver");
Connection con = DriverManager.getConnection("jdbc:mysql://172.16.1.25/songod", "root", "123456");
ResultSet rs = con.createStatement().executeQuery("select * from sod_song_ksc limit 20000");
ResultSetMetaData metadata = rs.getMetaData();
int columnCount = metadata.getColumnCount();
while (rs.next()) {
Map<String, Object> object = new HashMap<String, Object>();
for (int i = 1; i <= columnCount; i++) {
object.put(metadata.getColumnName(i), rs.getObject(i));
}
data.add(object);
}
rs.close();
con.close();
return data;
}
private static void index(List<Map<String, ?>> data) throws UnknownHostException {
Client client = getClient();
for (Map<String, ?> map : data) {
client.prepareIndex("songod", "sod_song_ksc").setSource(map).setId(map.get("SongID").toString())
.setRouting(map.get("SongID").toString()).execute().actionGet();
}
client.close();
}
private static void bulk(List<Map<String, ?>> data) throws UnknownHostException {
Client client = getClient();
BulkRequestBuilder bulkRequest = client.prepareBulk();
for (Map<String, ?> map : data) {
bulkRequest.add(client.prepareIndex("songod", "sod_song_ksc").setSource(map)
.setRouting(map.get("SongID").toString()).setId(map.get("SongID").toString()));
}
BulkResponse bulkResponse = bulkRequest.get();
client.close();
}
private static void get(List<Map<String, ?>> data) throws UnknownHostException {
Client client = getClient();
for (Map<String, ?> map : data) {
GetResponse response = client.prepareGet("songod", "sod_song_ksc", map.get("SongID").toString())
.setRouting(map.get("SongID").toString()).get();
String str = response.getSourceAsString();
// System.out.println(str);
}
client.close();
}
private static void delete(List<Map<String, ?>> data) throws UnknownHostException {
Client client = getClient();
for (Map<String, ?> map : data) {
client.prepareDelete("songod", "sod_song_ksc", map.get("SongID").toString())
.setRouting(map.get("SongID").toString()).get();
}
client.close();
}
private static void update(List<Map<String, ?>> data) throws IOException {
Client client = getClient();
for (Map<String, ?> map : data) {
client.prepareUpdate().setIndex("songod").setType("sod_song_ksc")
.setId(map.get("SongID").toString()).setDoc(jsonBuilder().startObject()
.field("Name", "Hello World!!" + map.get("Name").toString()).endObject())
.setRouting(map.get("SongID").toString()).get();
}
client.close();
}
private static void bulkProcessor(List<Map<String, ?>> data) throws IOException {
BulkProcessor bulkProcessor = BulkProcessor.builder(getClient(), new BulkProcessor.Listener() {
public void afterBulk(long arg0, BulkRequest arg1, BulkResponse arg2) {
// TODO Auto-generated method stub
}
public void afterBulk(long arg0, BulkRequest arg1, Throwable arg2) {
// TODO Auto-generated method stub
}
public void beforeBulk(long arg0, BulkRequest arg1) {
// TODO Auto-generated method stub
}
}).setBulkActions(1000).setBulkSize(new ByteSizeValue(3, ByteSizeUnit.MB))
.setFlushInterval(TimeValue.timeValueSeconds(5)).setConcurrentRequests(1)
.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)).build();
// .setRouting(map.get("SongID").toString())
for (Map<String, ?> map : data) {
UpdateRequest updateRequest = new UpdateRequest("songod", "sod_song_ksc", map.get("SongID").toString());
updateRequest.routing(map.get("SongID").toString());
updateRequest.id(map.get("SongID").toString());
updateRequest.doc(jsonBuilder().startObject().field("Name", "Hello World!!" + map.get("Name").toString())
.endObject());
bulkProcessor.add(updateRequest);
}
bulkProcessor.close();
}
private static Client getClient() throws UnknownHostException {
Settings settings = Settings.settingsBuilder().put("cluster.name", "es_cluster").build();
Client client = TransportClient.builder().settings(settings).build()
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("192.168.16.114"), 9300));
return client;
}
}

准备数据用时:1秒
单条索引:1381秒
单条更新:1624秒
单条删除:1393秒
批量索引:4秒
ID查询:8秒
bulkProcessor处理:10秒

运行环境是开发机.
不过可以看到,单条处理每秒也就是10-20条左右.
而在批量处理的时候,甚至可以达到每秒索引5000条.
所以批量处理几乎是ES的不二使用方法.

ITPub博客

ElasticSearch Java API

最新文章