ElasticSerarch bulk

蔚落 2022-05-15 08:54 150阅读 0赞

package com.Dao;  
import java.sql.Connection;  
import java.sql.PreparedStatement;  
import java.sql.ResultSet;  
import org.elasticsearch.action.bulk.BulkRequestBuilder;  
import org.elasticsearch.action.bulk.BulkResponse;  
import org.elasticsearch.client.Client;  
import org.elasticsearch.common.xcontent.XContentBuilder;  
import org.elasticsearch.common.xcontent.XContentFactory;  
import com.DB.DBPool;  
public class IndexData \{  
/\*\*  
\* 索引数据到ES,本文数据来自mysql  
\* @param indices  索引名称  
\* @param mappingType  索引类型  
\*/  
public static void createDate(String indices,String mappingType)\{

> > int temp=5000; //每次提交5000条记录
> 
> > Client client=ServerClient.getClient();
> 
> > Connection con =DBPool.getCn();
> 
> > BulkRequestBuilder bulkRequest=client.prepareBulk();  //容量，一次可索引几千乃至更多数据，根据各自电脑配置
> 
> > String sql=" select \* from class2 ";
> 
> > try \{ 
> 
> > > PreparedStatement p=con.prepareStatement(sql);
> 
> > > ResultSet query = p.executeQuery();
> 
> > > while(query.next())\{ 
> 
> > > > int i=query.getInt("id");  //此id连续自增，如不是连续，可设置变量代替
> 
> > > > XContentBuilder source=new XContentFactory().jsonBuilder()
> 
> > > > .startObject()
> 
> > > > > .field("id",query.getInt("id"))
> 
> > > > > .field("kw",query.getString("cname")==null?"":query.getString("cname"))
> 
> > > > > .field("edate", query.getDate("indate")==null?"":query.getDate("indate"))
> 
> > > > .endObject();
> 
> > > > //client.prepareIndex(indices,mappingType).setSource(source).execute().actionGet(); 一次索引一条数据，非常慢
> 
> > > > bulkRequest.add(client.prepareIndex(indices,mappingType,String.valueOf(i)).setSource(source));
> 
> > > > if(i%temp==0)\{ 
> 
> > > > BulkResponse response=bulkRequest.execute().actionGet();
> > > > 
> > > > bulkRequest=client.prepareBulk(); //提交完一次要从新new一次，不然下次提交的是temp的N倍数据  
> > > > 
> 
> > > > > if(response.hasFailures())\{ 
> 
> > > > > > System.out.println(query.getInt("id")+"失败");
> 
> > > > > \}else\{ 
> 
> > > > > > System.out.println(query.getInt("id")+"成功");
> 
> > > > > \}
> 
> > > > \}
> 
> > > \}
> 
> > \} catch (Exception e) \{ 
> 
> > > > e.printStackTrace();
> 
> > \}finally\{ 
> 
> > > client.close();
> 
> > > DBPool.Close(con);  
> > > 
> 
> > \}

> \}  
>  public static void main(String arg0\[\])\{  
>

> > createDate("lianan","lianan");

> \}

说明：该代码并个完整，只是写出了bulk的用法，业务根据各自需求改。