solr DIH 知识梳理

solr DIH 知识梳理

web.xml中listener配置

<listener>
        <listener-class>org.apache.solr.handler.dataimport.scheduler.ApplicationListener</listener-class>
</listener>

配置文件dataimport.properties

#################################################
#                                               #
#       dataimport scheduler properties         #
#                                               #
#################################################
 
#  to sync or not to sync
#  1 - active; anything else - inactive
syncEnabled=1
 
#  which cores to schedule
#  in a multi-core environment you can decide which cores you want syncronized
#  leave empty or comment it out if using single-core deployment
syncCores=core1,core2
 
#  solr server name or IP address
#  [defaults to localhost if empty]
server=localhost
 
#  solr server port
#  [defaults to 80 if empty]
port=8080
 
#  application name/context
#  [defaults to current ServletContextListener's context (app) name]
webapp=solr
 
#  URL params [mandatory]
#  remainder of URL
params=/dataimport?command=delta-import&clean=false&commit=true
 
#  schedule interval
#  number of minutes between two runs
#  [defaults to 30 if empty]
interval=1
 
#  重做索引的时间间隔，单位分钟，默认7200，即5天; 
#  为空,为0,或者注释掉:表示永不重做索引
reBuildIndexInterval=7200
 
#  重做索引的参数
reBuildIndexParams=/dataimport?command=full-import&clean=true&commit=true
 
#  重做索引时间间隔的计时开始时间，第一次真正执行的时间=reBuildIndexBeginTime+reBuildIndexInterval*60*1000；
#  两种格式：2012-04-11 03:10:00 或者  03:10:00，后一种会自动补全日期部分为服务启动时的日期
reBuildIndexBeginTime=03:10:00

• interval增量索引的频率，每隔interval分钟就启动一次task

 timer.scheduleAtFixedRate(task, startTime, 60000 * interval);

• 关于reBuildIndexBeginTime,这里表现为fullImportStartTime
• 增量更新的请求参数params=/dataimport?command=delta-import&clean=false&commit=true

fullImportTimer.scheduleAtFixedRate(fullImportTask, fullImportStartTime, 60000 * reBuildIndexInterval);

data-config.xml配置

• query是获取全部数据的SQL(solr从sql中获取那些数据)，多列

• deltaImportQuery是获取增量数据时使用的SQL(数据库新增数据追加到solr的数据)，多列

• deltaQuery是获取pk的SQL(数据库新增数据是,追加到solr的数据时的条件,根据id
  ,条件是最后一次获取的时间,${dataimporter.last_index_time,最后获取的时间})，一列

• 这个是在增量时使用的修改语句，其中需要注意的是dataimporter.delta这个前缀一定要带
  pk,根据我艰苦卓绝的跟踪代码知道这个pk其实作用只是用来对deltaQuery查询出来的内容放入到一个map中的时候作为key用的
  如果你不想deltaQuery查询出来的结果最后出现混乱，那么最好保证pk是唯一的

document中使用的是自己要被加入到索引中的field
query,被用来做为全量导入的时候使用
deltaImportQuery 这个是在增量时使用的修改语句，其中需要注意的是dataimporter.delta这个前缀一定要带
pk,根据我艰苦卓绝的跟踪代码知道这个pk其实作用只是用来对deltaQuery查询出来的内容放入到一个map中的时候作为key用的
如果你不想deltaQuery查询出来的结果最后出现混乱，那么最好保证pk是唯一的
deltaQuery，这个是用来查询需要被更新的对象的主键，一边deltaImportQuery使用
transformer：很多时候数据库中的字段不能满足你的需要，比如存储了用户生日，那么你需要将他的生肖存储，则此时需要对生日做自己的处理

增量索引

终止跑索引：http://localhost:8080/solr/collection1/dataimport?command=abort
开始索引：http://localhost:8080/solr/collection1/dataimport?command=full-import
增量索引 ：http://localhost:8080/solr/collection1/dataimport?command=delta-import

源代码apache-solr-dataimportscheduler-1.4.jar
ApplicationListener

package org.apache.solr.handler.dataimport.scheduler;

import java.util.Calendar;
import java.util.Date;
import java.util.Timer;
import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ApplicationListener
  implements ServletContextListener
{
  private static final Logger logger = LoggerFactory.getLogger(ApplicationListener.class)
    ;

  public void contextDestroyed(ServletContextEvent servletContextEvent)
  {
    ServletContext servletContext = servletContextEvent.getServletContext();

    Timer timer = (Timer)servletContext.getAttribute("timer");

    Timer fullImportTimer = (Timer)servletContext
      .getAttribute("fullImportTimer");

    if (timer != null)
      timer.cancel();
    if (fullImportTimer != null) {
      fullImportTimer.cancel();
    }

    servletContext.removeAttribute("timer");
    servletContext.removeAttribute("fullImportTimer");
  }

  public void contextInitialized(ServletContextEvent servletContextEvent)
  {
    ServletContext servletContext = servletContextEvent.getServletContext();
    try
    {
      Timer timer = new Timer();
    //增量索引task
      DeltaImportHTTPPostScheduler task = new DeltaImportHTTPPostScheduler(servletContext
        .getServletContextName(), timer);
    //配置的间隔时间分钟
      int interval = task.getIntervalInt();

      Calendar calendar = Calendar.getInstance();

      calendar.add(12, interval);
      Date startTime = calendar.getTime();
      //task调度
      timer.scheduleAtFixedRate(task, startTime, 60000 * interval);

      servletContext.setAttribute("timer", timer);

      Timer fullImportTimer = new Timer();
      //全量索引task
      FullImportHTTPPostScheduler fullImportTask = new FullImportHTTPPostScheduler(servletContext
        .getServletContextName(), fullImportTimer);
      //这里重建索引时间
      int reBuildIndexInterval = fullImportTask
        .getReBuildIndexIntervalInt();
      if (reBuildIndexInterval <= 0) {
        logger.warn("Full Import Schedule disabled");
        return;
      }

      Calendar fullImportCalendar = Calendar.getInstance();
      Date beginDate = fullImportTask.getReBuildIndexBeginTime();
      fullImportCalendar.setTime(beginDate);
      fullImportCalendar.add(12, reBuildIndexInterval);
      Date fullImportStartTime = fullImportCalendar.getTime();
      //fullImportStartTime这个跟配置文件里的reBuildIndexBeginTime相关，
      fullImportTimer.scheduleAtFixedRate(fullImportTask, fullImportStartTime, 60000 * reBuildIndexInterval);

      servletContext.setAttribute("fullImportTimer", fullImportTimer);
    }
    catch (Exception e) {
      if (e.getMessage().endsWith("disabled"))
        logger.warn("Schedule disabled");
      else
        logger.error("Problem initializing the scheduled task: ", e);
    }
  }
}

增量，全量索引task都继承自BaseTimerTask，主要都差不多，看BaseTimerTask的prepUrlSendHttpPost就好
增量索引task

package org.apache.solr.handler.dataimport.scheduler;

import java.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DeltaImportHTTPPostScheduler extends BaseTimerTask
{
  private static final Logger logger = LoggerFactory.getLogger(DeltaImportHTTPPostScheduler.class)
    ;

  public DeltaImportHTTPPostScheduler(String webAppName, Timer t) throws Exception
  {
    super(webAppName, t);
    logger.info("<index update process> DeltaImportHTTPPostScheduler init");
  }

  public void run()
  {
    try {
      if ((this.server.isEmpty()) || (this.webapp.isEmpty()) || (this.params == null) || 
        (this.params
        .isEmpty())) {
        logger.warn("<index update process> Insuficient info provided for data import");
        logger.info("<index update process> Reloading global dataimport.properties");
        reloadParams();
      }
      else if (this.singleCore) {
        prepUrlSendHttpPost(this.params);
      }
      else if ((this.syncCores.length == 0) || ((this.syncCores.length == 1) && 
        (this.syncCores[0]
        .isEmpty()))) {
        logger.warn("<index update process> No cores scheduled for data import");
        logger.info("<index update process> Reloading global dataimport.properties");
        reloadParams();
      }
      else {
        for (String core : this.syncCores)
          prepUrlSendHttpPost(core, this.params);
      }
    }
    catch (Exception e) {
      logger.error("Failed to prepare for sendHttpPost", e);
      reloadParams();
    }
  }
}

全量索引task

package org.apache.solr.handler.dataimport.scheduler;

import java.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class FullImportHTTPPostScheduler extends BaseTimerTask
{
  private static final Logger logger = LoggerFactory.getLogger(FullImportHTTPPostScheduler.class)
    ;

  public FullImportHTTPPostScheduler(String webAppName, Timer t) throws Exception
  {
    super(webAppName, t);
    logger.info("<index update process> DeltaImportHTTPPostScheduler init");
  }

  public void run()
  {
    try {
      if ((this.server.isEmpty()) || (this.webapp.isEmpty()) || (this.reBuildIndexParams == null) || 
        (this.reBuildIndexParams
        .isEmpty())) {
        logger.warn("<index update process> Insuficient info provided for data import, reBuildIndexParams is null");
        logger.info("<index update process> Reloading global dataimport.properties");
        reloadParams();
      }
      else if (this.singleCore) {
        prepUrlSendHttpPost(this.reBuildIndexParams);
      }
      else if ((this.syncCores.length == 0) || ((this.syncCores.length == 1) && 
        (this.syncCores[0]
        .isEmpty()))) {
        logger.warn("<index update process> No cores scheduled for data import");
        logger.info("<index update process> Reloading global dataimport.properties");
        reloadParams();
      }
      else {
        for (String core : this.syncCores)
          prepUrlSendHttpPost(core, this.reBuildIndexParams);
      }
    }
    catch (Exception e) {
      logger.error("Failed to prepare for sendHttpPost", e);
      reloadParams();
    }
  }
}

BaseTimerTask的，这里主要关注prepUrlSendHttpPost

package org.apache.solr.handler.dataimport.scheduler;

import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class BaseTimerTask extends TimerTask
{
  protected String syncEnabled;
  protected String[] syncCores;
  protected String server;
  protected String port;
  protected String webapp;
  protected String params;
  protected String interval;
  protected String cores;
  protected SolrDataImportProperties p;
  protected boolean singleCore;
  protected String reBuildIndexParams;
  protected String reBuildIndexBeginTime;
  protected String reBuildIndexInterval;
  private String webAppName;
  protected static final Logger logger = LoggerFactory.getLogger(BaseTimerTask.class)
    ;

  public BaseTimerTask(String webAppName, Timer t) throws Exception
  {
    this.webAppName = webAppName;

    this.p = new SolrDataImportProperties();
    reloadParams();

    if (!this.syncEnabled.equals("1")) {
      throw new Exception("Schedule disabled");
    }
    if ((this.syncCores == null) || ((this.syncCores.length == 1) && 
      (this.syncCores[0]
      .isEmpty()))) {
      this.singleCore = true;
      logger.info("<index update process> Single core identified in dataimport.properties");
    } else {
      this.singleCore = false;
      logger.info(new StringBuilder().append("<index update process> Multiple cores identified in dataimport.properties. Sync active for: ").append(this.cores).toString());
    }
  }

  protected void reloadParams()
  {
    this.p.loadProperties(true);
    this.syncEnabled = this.p.getProperty("syncEnabled");
    this.cores = this.p.getProperty("syncCores");
    this.server = this.p.getProperty("server");
    this.port = this.p.getProperty("port");
    this.webapp = this.p.getProperty("webapp");
    this.params = this.p.getProperty("params");
    this.interval = this.p.getProperty("interval");
    this.syncCores = (this.cores != null ? this.cores.split(",") : null);

    this.reBuildIndexParams = this.p
      .getProperty("reBuildIndexParams");

    this.reBuildIndexBeginTime = this.p
      .getProperty("reBuildIndexBeginTime");

    this.reBuildIndexInterval = this.p
      .getProperty("reBuildIndexInterval");

    fixParams(this.webAppName);
  }

  protected void fixParams(String webAppName) {
    if ((this.server == null) || (this.server.isEmpty()))
      this.server = "localhost";
    if ((this.port == null) || (this.port.isEmpty()))
      this.port = "8080";
    if ((this.webapp == null) || (this.webapp.isEmpty()))
      this.webapp = webAppName;
    if ((this.interval == null) || (this.interval.isEmpty()) || (getIntervalInt() <= 0))
      this.interval = "30";
    if ((this.reBuildIndexBeginTime == null) || (this.reBuildIndexBeginTime.isEmpty()))
      this.interval = "00:00:00";
    if ((this.reBuildIndexInterval == null) || (this.reBuildIndexInterval.isEmpty()) || 
      (getReBuildIndexIntervalInt() <= 0))
      this.reBuildIndexInterval = "0";
  }

  protected void prepUrlSendHttpPost(String params)
  {
    sendHttpPost(null, params);
  }

  protected void prepUrlSendHttpPost(String coreName, String params)
  {
    sendHttpPost(coreName, params);
  }

  protected void sendHttpPost(String coreName, String params)
  {
    DateFormat df = new SimpleDateFormat("dd.MM.yyyy HH:mm:ss SSS");
    Date startTime = new Date();

    String core = coreName == null ? "" : new StringBuilder().append("[").append(coreName).append("] ").toString();

    logger.info(new StringBuilder().append(core).append("<index update process> Process started at .............. ")
      .append(df
      .format(startTime))
      .toString());
    try
    {
      String completeUrl = buildUrl(coreName, params);
      URL url = new URL(completeUrl);
      HttpURLConnection conn = (HttpURLConnection)url.openConnection();

      conn.setRequestMethod("POST");
      conn.setRequestProperty("type", "submit");
      conn.setDoOutput(true);

      conn.connect();

      logger.info(new StringBuilder().append(core).append("<index update process> Full URL				").append(conn.getURL()).toString());
      logger.info(new StringBuilder().append(core).append("<index update process> Response message			")
        .append(conn
        .getResponseMessage()).toString());
      logger.info(new StringBuilder().append(core).append("<index update process> Response code			")
        .append(conn
        .getResponseCode()).toString());

      if (conn.getResponseCode() != 200) {
        reloadParams();
      }

      conn.disconnect();
      logger.info(new StringBuilder().append(core).append("<index update process> Disconnected from server		").append(this.server).toString());
      Date endTime = new Date();
      logger.info(new StringBuilder().append(core).append("<index update process> Process ended at ................ ")
        .append(df
        .format(endTime))
        .toString());
    } catch (MalformedURLException mue) {
      logger.error("Failed to assemble URL for HTTP POST", mue);
    } catch (IOException ioe) {
      logger.error("Failed to connect to the specified URL while trying to send HTTP POST", ioe);
    }
    catch (Exception e)
    {
      logger.error("Failed to send HTTP POST", e);
    }
  }

  private String buildUrl(String coreName, String params) { StringBuilder sb = new StringBuilder();

    sb.append("http://").append(this.server).append(":").append(this.port);

    if (!this.webapp.startsWith("/"))
      sb.append("/");
    sb.append(this.webapp);

    if ((coreName != null) && (!coreName.isEmpty())) {
      if (!this.webapp.endsWith("/"))
        sb.append("/");
      sb.append(coreName);
    }

    if (sb.charAt(sb.length() - 1) == '/') {
      if (params.startsWith("/"))
        sb.setLength(sb.length() - 1);
    }
    else if (!params.startsWith("/")) {
      sb.append("/");
    }
    sb.append(params);

    return sb.toString(); }

  public int getIntervalInt() {
    try {
      return Integer.parseInt(this.interval);
    } catch (NumberFormatException e) {
      logger.warn("Unable to convert 'interval' to number. Using default value (30) instead", e);
    }

    return 30;
  }

  public int getReBuildIndexIntervalInt()
  {
    try {
      return Integer.parseInt(this.reBuildIndexInterval);
    } catch (NumberFormatException e) {
      logger.info("Unable to convert 'reBuildIndexInterval' to number. do't rebuild index.", e);
    }

    return 0;
  }

  public Date getReBuildIndexBeginTime()
  {
    Date beginDate = null;
    try {
      SimpleDateFormat sdfDate = new SimpleDateFormat("yyyy-MM-dd");
      String dateStr = sdfDate.format(new Date());
      beginDate = sdfDate.parse(dateStr);
      if ((this.reBuildIndexBeginTime == null) || 
        (this.reBuildIndexBeginTime
        .isEmpty()))
        return beginDate;
      SimpleDateFormat sdf;
      if (this.reBuildIndexBeginTime.matches("\d{2}:\d{2}:\d{2}")) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        beginDate = sdf.parse(new StringBuilder().append(dateStr).append(" ").append(this.reBuildIndexBeginTime).toString());
      }
      else if (this.reBuildIndexBeginTime
        .matches("\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}"))
      {
        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
      }
      return sdf.parse(this.reBuildIndexBeginTime);
    }
    catch (ParseException e)
    {
      logger.warn("Unable to convert 'reBuildIndexBeginTime' to date. use now time.", e);
    }

    return beginDate;
  }
}

最后还有Properties文件相关的类SolrDataImportProperties

package org.apache.solr.handler.dataimport.scheduler;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Properties;
import org.apache.solr.core.SolrResourceLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SolrDataImportProperties
{
  private Properties properties;
  public static final String SYNC_ENABLED = "syncEnabled";
  public static final String SYNC_CORES = "syncCores";
  public static final String SERVER = "server";
  public static final String PORT = "port";
  public static final String WEBAPP = "webapp";
  public static final String PARAMS = "params";
  public static final String INTERVAL = "interval";
  public static final String REBUILDINDEXPARAMS = "reBuildIndexParams";
  public static final String REBUILDINDEXBEGINTIME = "reBuildIndexBeginTime";
  public static final String REBUILDINDEXINTERVAL = "reBuildIndexInterval";
  private static final Logger logger = LoggerFactory.getLogger(SolrDataImportProperties.class)
    ;

  public void loadProperties(boolean force)
  {
    try
    {
      SolrResourceLoader loader = new SolrResourceLoader(null);
      logger.info("Instance dir = " + loader.getInstanceDir());

      String configDir = loader.getConfigDir();
      configDir = SolrResourceLoader.normalizeDir(configDir);
      if ((force) || (this.properties == null)) {
        this.properties = new Properties();

        String dataImportPropertiesPath = configDir + "dataimport.properties";

        FileInputStream fis = new FileInputStream(dataImportPropertiesPath);

        this.properties.load(fis);
      }
    } catch (FileNotFoundException fnfe) {
      logger.error("Error locating DataImportScheduler dataimport.properties file", fnfe);
    }
    catch (IOException ioe)
    {
      logger.error("Error reading DataImportScheduler dataimport.properties file", ioe);
    }
    catch (Exception e)
    {
      logger.error("Error loading DataImportScheduler properties", e);
    }
  }

  public String getProperty(String key) {
    return this.properties.getProperty(key);
  }
}

服务端流程主要流程

handleRequestBody-->  importer.runCmd-->doFullImport/doDeltaImport--> docBuilder.execute()--> doDelta()-->collectDelta()-->ject> row = epw.nextModifiedRowKey()/getModifiedParentRows-->entityProcessor.nextModifiedParentRowKey()-->initQuery()-->dataSource.getData(q)-->ResultSetIterator r = new ResultSetIterator(query)-->getARow()

剩下的就跟正常的solr处理流程差不多了

links

Solr中DIH模式的使用

solr连接数据库配置

Solr学习（五）DIH增量、定时导入并检索数据