[关闭]
@zhou-si 2016-09-11T02:14:58.000000Z 字数 6878 阅读 2327

solr按小时分片抽取一天数据

solr

简介

最近几天项目组有个紧急任务：从solr中抽取数据入hive库，我也不清楚为何要有这种业务需求，其实solr里是可以配置数据存储路径的，直接指定为hdfs某个目录也很简单。由于项目紧我也只是把solr数据抽取到windows上然后传回hdfs，其实可以直接打成jar包指定写出路径的……只介绍代码的实现和所需要的jar包

环境和所需jar包

jdk8 apache-solr-solrj-3.4.0.jar commons-httpclient-3.1.jar commons-io-2.5.jar commons-lang-2.6.jar commons-logging-1.1.1.jar httpclient-4.4.1.jar httpcore-4.4.1.jar httpmime-4.2.3.jar noggit-0.5.jar slf4j-api-1.7.7.jar solr-solrj-6.0.0.jar zookeeper-3.4.6.jar

代码

  1. package org.apache.solr.test.base.service.impl;
  3. import java.io.File;
  4. import java.io.FileWriter;
  5. import java.util.Date;
  6. import java.util.HashSet;
  7. import java.util.List;
  8. import org.apache.commons.lang.StringUtils;
  9. import org.apache.commons.lang.time.DateFormatUtils;
  10. import org.apache.solr.client.solrj.SolrClient;
  11. import org.apache.solr.client.solrj.SolrQuery;
  12. import org.apache.solr.client.solrj.SolrRequest;
  13. import org.apache.solr.client.solrj.impl.CloudSolrClient;
  14. import org.apache.solr.client.solrj.impl.HttpSolrClient;
  15. import org.apache.solr.client.solrj.response.QueryResponse;
  16. import org.apache.solr.common.SolrDocument;
  17. import org.apache.solr.common.SolrDocumentList;
  18. import org.apache.solr.common.params.FacetParams;
  19. import org.apache.solr.common.params.GroupParams;
  20. import com.sun.org.apache.xerces.internal.impl.xpath.regex.ParseException;
  21. /**
  22.  * 创建时间：20160728
  23.  * 功能：抽取solr数据写到文件
  24.  * @author 圣斗士宙斯
  25.  *
  26.  */
  27. public abstract class BaseSolrServiceImpl_ec {
  28. protected static SolrClient client;
  29. /**
  30.  * 初始化solr连接
  31.  */
  32.     private static void init() {
  33.         try {
  34.             client = new CloudSolrClient(
  35.                     "zk_host1:port,zk_host2:port");//zookeeper的ip和端口 
  36.             ((CloudSolrClient) client).setDefaultCollection("xxx");
  37.             ((CloudSolrClient) client).connect();
  38.         } catch (Exception e) {
  39.             client = new HttpSolrClient("http://solr_ip:port/solr/xxx");//solr库的链接 
  40.         }
  41.     }
  42.     /**
  43.     *查询方法（可以不做排序过滤等）
  44.     */
  45.     public static QueryResponse query(String query, String fields,
  46.             int pageNumber, int size, List<SolrQuery.SortClause> fieldSorts,
  47.             String... conditions) throws Exception {
  49.         SolrQuery solrQuery = new SolrQuery();
  50.         //以下都是些控制条件，不用也可以删掉
  51.         if (StringUtils.isBlank(query)) {
  52.             solrQuery.setQuery("*:*");
  53.             // solrQuery.addFacetQuery("*:*");无条件全表扫
  54.         } else {
  55.             solrQuery.setQuery(query);
  56.         }
  57.         size = size <= 0 ? 10 : size;
  58.         pageNumber = pageNumber < 1 ? 0 : (pageNumber - 1) * size;
  59.         solrQuery.setStart(pageNumber);
  60.         solrQuery.setRows(size);
  61.         if (StringUtils.isNotBlank(fields)) {
  62.             solrQuery.setFields(fields);
  63.         }
  64.         if (fieldSorts != null) {
  65.             for (SolrQuery.SortClause fieldSort : fieldSorts) {
  66.                 solrQuery.addSort(fieldSort);
  67.             }
  68.         }
  69.         if (conditions != null) {
  70.             if (conditions.length >= 1 && StringUtils.isNotBlank(conditions[0])) {
  71.                 //solrQuery.setFilterQueries(conditions[0].split(",|，"));
  72.                 solrQuery.add("shards", conditions[0]);
  73.             }
  74.             if (conditions.length >= 2 && StringUtils.isNotBlank(conditions[1])) {
  75.                 String[] split = conditions[1].split(",|，");
  76.                 solrQuery.setParam("group", true);
  77.                 solrQuery.setParam("group.ngroups", true);
  78.                 solrQuery.setParam("group.field", split);
  79.                 solrQuery.setParam(GroupParams.GROUP_SORT, "pub_time desc");
  80.             }
  81.             if (conditions.length >= 3 && StringUtils.isNotBlank(conditions[2])) {
  82.                 String[] split = conditions[2].split(";|；");
  83.                 solrQuery.set(FacetParams.FACET, true);
  84.                 if (split.length > 1) {
  85.                     solrQuery.set("facet.date", split[0]);
  86.                     solrQuery.set("facet.date.start", split[1]);
  87.                     solrQuery.set("facet.date.end", split[2]);
  88.                     solrQuery.set("facet.date.gap", split[3]);
  89.                 } else {
  90.                     solrQuery.set("facet.field", split[0]);
  91.                 }
  92.             }
  93.             if (conditions.length >= 5 && StringUtils.isNotBlank(conditions[4])) {
  94.                 solrQuery.setHighlight(true);
  95.                 solrQuery.setParam("hl.fl", conditions[4]);
  96.                 solrQuery.setParam("hl.fragsize", "160");
  97.                 solrQuery.setParam("hl.snippets", "2");
  98.                 solrQuery.setParam("hl.mergeContiguous", "true");
  99.                 solrQuery
  100.                         .setHighlightSimplePre("<font class='word-highlight'>");
  101.                 solrQuery.setHighlightSimplePost("</font>");
  102.             }
  103.             if (conditions.length >= 6 && StringUtils.isNotBlank(conditions[5])) {
  104.                 solrQuery.setParam("hl.q", conditions[5]);
  105.             }
  106.         }
  107.         if (client instanceof CloudSolrClient) {
  108.             if (conditions.length >= 4 && StringUtils.isNotBlank(conditions[3])) {
  109.                 solrQuery.add("shards", conditions[3]);
  110.             }
  111.         }
  112.         QueryResponse queryResponse = client.query(solrQuery,
  113.                 SolrRequest.METHOD.POST);
  114.         return queryResponse;
  115.     }
  117.     public static void main(String[] args) {
  118.         //调用初始化连接方法
  119.         init();
  120.         //接收传入的参数，这里只是对时间做限制，一天内24小时
  121.         String whereColumn = args[0];// crawltime
  122.         String time1 = args[1];// 2016-01-17
  123.         String time2 = time1;
  124.         String partitions = args[2];
  125.         //由于按天分页抽取solr库的数据会非常慢，分为24此抽取，分页数按照当前小时的总数量/一页显示的数量
  126.         try {
  127.             QueryResponse response = null;
  128.             FileWriter fw = null;// BufferedWriter FileWriter
  129.             StringBuilder sb = new StringBuilder();// StringBuffer
  130.             for (int i = 0; i <= 23; i++) {
  131.                 // String stringHour ;
  132.                 if (i <= 9) {
  133.                     // stringHour = "0"+(i+=1);
  134.                     time1 = time1 + "T0" + i + ":00:00Z";
  135.                     if (i+1 == 10) {
  136.                         time2 = time2 + "T" + (i+1) + ":00:00Z";
  137.                     }else {
  138.                         time2 = time2 + "T0" + (i+1) + ":00:00Z";
  139.                     }
  140.                 } else {
  141.                     // stringHour = (i+=1)+"";
  142.                     time1 = time1 + "T" + i + ":00:00Z";
  143.                     time2 = time2 + "T" + (i+1) + ":00:00Z";
  144.                 }
  145.                 response = query(whereColumn + ":[" + time1 + " TO "
  146.                         + time2 + "] && title:*手机*"
  147.                         + " || ( weburl:*mobile.zol.com.cn* || weburl:*mo*)", null,
  148.                         0, 5000, null,partitions);
  149.                 int numFound = (int) response.getResults().getNumFound();
  150.                 int fors = (int) Math.ceil((numFound / 5000)); 
  151.                 String string = time1.substring(0, 10).replace("-", "");
  152.                 System.out.println("第 "+time1 +"小时到第"+time2+"小时 统计条数: "+numFound);// 统计条数
  153.                 System.out.println("第 "+time1 +"小时到第"+time2+"小时 分页总数: "+(int) Math.ceil((numFound / 5000)));// 分页总数
  155.                 File file_Dir = new File("C:\\Users\\zhuqitian\\Desktop\\solrDate_ec\\"+string+"\\");
  156.                 if (!file_Dir.exists()) {
  157.                     file_Dir.mkdirs();
  158.                 }
  159.                 File file = new File("C:\\Users\\zhuqitian\\Desktop\\solrDate_ec\\"+string+"\\solrDate_ec_"
  160.                         + string + "_"+i+".txt");
  161.                 if (!file.exists()) {
  162.                     file.createNewFile();
  163.                 }
  164.                 fw = new FileWriter(file, true);// FileOutputStream
  165.                 for (int k = 0; k <= fors; k++) {
  166.                     SolrDocumentList docs1 = response.getResults();
  167.                     response = query(whereColumn + ":[" + time1 + " TO "
  168.                             + time2 + "]", null, k, 5000, null);
  169.                     for (SolrDocument sd : docs1) {
  170.                         sb.append(sd.getFieldValue("doc_id") + "!@#");
  171.                         sb.append(sd.getFieldValue("dedup_id") + "!@#");
  172.                         sb.append(sd.getFieldValue("url") + "!@#");
  173.                         sb.append(sd.getFieldValue("author") + "!@#");
  174.                         sb.append(sd.getFieldValue("weburl") + "!@#");
  175.                         sb.append(sd.getFieldValue("media") + "!@#");
  176.                         sb.append(sd.getFieldValue("region") + "!@#");
  177.                         sb.append(sd.getFieldValue("source") + "!@#");
  178.                         sb.append(sd.getFieldValue("create_time") + "!@#");
  179.                         sb.append(sd.getFieldValue("summary") + "!@#");
  180.                         sb.append(sd.getFieldValue("keywords") + "!@#");
  181.                         sb.append(sd.getFieldValue("simple_keywords") + "!@#");
  182.                         sb.append(sd.getFieldValue("sentiment") + "!@#");
  183.                         sb.append(sd.getFieldValue("sentiment_val") + "!@#");
  184.                         sb.append(sd.getFieldValue("comment_count") + "!@#");
  185.                         sb.append(sd.getFieldValue("click_count") + "!@#");
  186.                         sb.append(sd.getFieldValue("crawltime") + "!@#");
  187.                         sb.append(sd.getFieldValue("title") + "!@#");
  188.                         sb.append(sd.getFieldValue("pub_time") + "!@#");
  189.                         sb.append("\n");
  190.                         // bytes = sb.toString().getBytes();
  191.                     }
  192.                     fw.write(sb.toString());
  193.                     System.out.println("第 "+ i +" 小时到第 "+(i+1)+"小时： 第 " + k +" 页写入完毕！文件名为 " + file.getName());
  194.                     sb.delete(0, sb.length());
  195.                 }
  196.                 time1 = args[1];
  197.                 time2 = time1;
  198.             }
  199.             fw.close();
  200.         } catch (Exception e) {
  201.             e.printStackTrace();
  202.         }
  204.     }
  206. }

效果

startTime: 2016-05-15T00:00:00Z
endTime: 2016-05-15T01:00:00Z
统计条数: 62308
分页总数: 12
startTime: 2016-05-15T01:00:00Z
endTime: 2016-05-15T02:00:00Z
统计条数: 41098
分页总数: 8
…………
第 0 小时到第 1小时： 第 0 页写入完毕！文件名为solrDate_ec_20160515_0.txt
第 0 小时到第 1小时： 第 1 页写入完毕！文件名为solrDate_ec_20160515_0.txt
第 0 小时到第 1小时： 第 2 页写入完毕！文件名为solrDate_ec_20160515_0.txt
…………
执行文件夹中会列出每个小时写进来的文件
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注