使用Camel在来自不同来源的Solr中索引数据

Apache Solr是建立在Lucene之上的“流行的，快速的开源企业搜索平台”。为了进行搜索（并查找结果），通常需要从不同的来源（例如内容管理系统，关系数据库，旧系统）中提取数据，这是您最初的要求……然后，还要保持索引的最高难度。通过添加新数据，更新现有记录，删除过时数据来确定日期。新数据源可能与初始数据源相同，但也可能是Twitter，AWS或其余端点之类的数据源。

Solr可以理解不同的文件格式，并提供大量的数据选项
索引：

直接HTTP和远程流传输–通过发布用于直接索引的文件或用于远程流传输的文件路径，允许您通过HTTP与Solr进行交互。
DataImportHandler –是一个模块，可从关系数据库或文件系统中同时导入增量增量和增量增量。
SolrJ –一个使用Apache Commons HTTP Client访问Solr的Java客户端。

但是在现实生活中，用数百万个文档，数十个转换，过滤，内容丰富，复制，并行处理来索引来自不同来源的数据不仅需要更多。解决这种挑战的一种方法是重新发明轮子：编写少量自定义应用程序，将它们与一些脚本结合起来或运行cronjobs。另一种方法是使用一种灵活的工具，该工具设计为可配置和可插入的，可以帮助您轻松扩展和分配负载。这样的工具是Apache Camel，它现在也具有Solr 连接器。

一切始于几个月前，在Sourcesense的大本营期间，我和我的同事Alex在这里尝试不同的项目，以实现将数据索引到Solr中的管道。不出所料，我们发现了Camel，经过几天的配对，我们已经准备好将最初的Solr组件版本提交给Camel，并由Ben Oday进行了进一步扩展。目前，它具有功能齐全的Solr连接器，该连接器在后台使用SolrJ，并允许您：配置SolrServer和StreamingUpdateSolrServer的所有参数；支持以下操作：insert，add_bean，delete_by_id，delete_by_query，commit，rolback，optimize；索引文件，SolrInputDocument实例，具有批注或单个消息头的Bean。

创建一个骆驼路线来索引关系数据库表和本地文件系统中的所有数据很简单：

public void configure() {
from("timer://clear?repeatCount=1").to("direct:clearIndex");from("file:src/data?noop=true").to("direct:insert");from("timer://database?repeatCount=1").to("sql:select * from products?dataSourceRef=productDataSource").split(body()).process(new SqlToSolrMapper()).to("direct:insert");from("direct:insert").setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_INSERT)).to(SOLR_URL).setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_COMMIT)).to(SOLR_URL);from("direct:clearIndex").setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_DELETE_BY_QUERY)).setBody(constant("*:*")).to(SOLR_URL).setHeader(SolrConstants.OPERATION, constant(SolrConstants.OPERATION_COMMIT)).to(SOLR_URL);
}

上面的方法将首先删除所有文档，然后再删除提交，以清除索引。然后它将开始从src / data文件夹轮询文件，读取每个文件并将其发送到Solr端点。假设文件采用Solr可以理解的格式，则将对它们进行索引并提交。第三条路线将从数据库（在内存中）检索所有产品，将它们拆分成单独的记录，将每个记录映射到Solr字段，然后进行摘要。幸运的是，在2012年，软件开发人员的生活并不那么简单。如今，取而代之的是，更实际的索引要求将由以下内容组成：

如果文件获得批准，请尽快提交，否则每10分钟提交一次。

骆驼如何为您提供帮助？骆驼支持包括S3在内的大多数流行的Amazon API。使用aws-s3组件，可以从S3存储桶中读取文件，然后对批准的文档应用过滤器，以便将它们发送到单独的路由中以进行即时提交。

<route><from uri="aws-s3://MyBucket?delay=5000&maxMessagesPerPoll=5"/><choice><when><xpath>/add/doc[@status='approved']</xpath><to uri="direct:indexAndCommit"/></when><otherwise><to uri="direct:index"/></otherwise></choice>
</route>
<route><from uri="timer://commit?fixedRate=true&period=600s"/><from uri="direct:commit"/>
</route>

通过调用XXX外部服务来丰富经纬度的地址数据，以方便在Solr中进行空间搜索。

<route id="fromDB"><from uri="jpa://com.ofbizian.pipeline.Customer?consumer.namedQuery= newCustomers&maximumResults=10&delay=5000"/><enrich uri="direct:coordinateEnricher" strategyRef="latLongAggregationStrategy"/><to uri="direct:index"/></route><route><from uri="direct:coordinateEnricher"/><setHeader headerName="CamelHttpQuery"><simple>address='${body.address}'&sensor=false</simple></setHeader><to uri="http://maps.google.com/maps/api/geocode/xml"/><setHeader headerName="lat"><xpath resultType="java.lang.Double">//result[1]/geometry/location/lat/text()</xpath></setHeader><setHeader headerName="lng"><xpath resultType="java.lang.Double">//result[1]/geometry/location/lng/text()</xpath></setHeader></route>

上面的路线一次从“客户”表10记录中读取数据，并且每条路线都将使用客户地址字段调用google的maps API来获取经度和纬度。使用XPath从响应中提取坐标，然后将其合并回Customer对象。很简单，不是吗。

3.在我们的内容管理系统中，此/那个/路径下的内容编入索引，并监视更新。

<route><from uri="jcr://user:pass@repository/import/inbox/signal?eventTypes=3&deep=true&synchronous=false"/><to uri="direct:index"/>
</route>

Camel有一个jcr连接器，使您可以在任何Java内容存储库中创建内容。 CAMEL-5155中还提交了一项改进，该改进将允许很快从JCR v.2支持存储库中读取内容。如果幸运的话，并且您的CMS支持CMIS，则可以出于相同目的使用来自github的camel-cmis连接器。

4.聆听有关我们产品/公司的推文，进行情感分析，并仅索引积极的推文。

<route id="fromTwitter"><from uri="twitter://streaming/filter?type=event&keywords=productName&consumerKey={{consumer.key}}&consumerSecret={{consumer.secret}}"/><setHeader headerName="CamelHttpQuery"><language language="beanshell">"q=" + java.net.URLEncoder.encode(request.getBody().getText(), "UTF-8")</language></setHeader><throttle timePeriodMillis="1500"><constant>1</constant><to uri="http://data.tweetsentiments.com:8080/api/analyze.xml"/><setHeader headerName="sentiment"><xpath resultType="java.lang.Double">/sentiment/value/text()</xpath></setHeader><filter><simple>${in.header.sentiment} > 0</simple><to uri="direct:index"/></filter></throttle>
</route>

这条路线将使用Twitter的实时api侦听tweet，对tweet进行url编码，并调用tweetsentiments API进行情感分析。此外，它将应用限制，因此每秒钟调用次数最多受限制，因此每1500毫秒最多仅发出一个请求。然后，该路由将在应用索引之前应用过滤器以忽略所有否定推文。

如您所见，Camel可以轻松地与许多不同的系统（包括Solr）进行交互，即使您有非常自定义的应用程序，编写连接器也并不困难。但这只是故事的一方面。另一方面，还有Camel实施的企业集成模式的完整列表，这些通道对于任何严重的数据摄取管道都是必需的：路由器，转换器，过滤器，拆分器，聚合器，Content Enricher，负载均衡器…最后但并非最不重要的：异常处理，日志记录，监视，DSL……两个词： 骆驼岩！

PS ：示例的完整源代码可以在我的github帐户上找到。

参考： 使用来自OFBIZian博客的JCG合作伙伴 Bilgin Ibryam的Camel在来自不同来源的Solr中建立索引数据。