Изучите Apache Camel: индексирование твитов в режиме реального времени

В большинстве проектов по разработке программного обеспечения есть момент, когда приложение должно начать взаимодействовать с другими приложениями или сторонними компонентами.

Будь то отправка уведомления по электронной почте, вызов внешнего API, запись в файл или перенос данных из одного места в другое, вы либо развертываете собственное решение, либо используете существующую среду.

Что касается существующих сред в экосистеме Java, на одном конце спектра мы находим  TIBCO BusinessWorks  и Mule ESB , а на другом конце — Spring Integration  и Apache Camel .

В этом уроке я собираюсь познакомить вас с Apache Camel с  помощью примера приложения, которое считывает твиты из примера ленты Twitter и индексирует эти твиты в реальном времени с помощью  ElasticSearch .

Что такое Apache Camel?

Интеграция приложения с внутренними или внешними компонентами в экосистеме является одной из самых сложных задач в разработке программного обеспечения, и если она не сделана правильно, это может привести к огромному беспорядку и реальной боли в обслуживании в долгосрочной перспективе.

К счастью, Camel — интегрированная среда с открытым исходным кодом, размещенная в Apache, — основана на корпоративных  шаблонах интеграции, и эти шаблоны могут помочь в написании более удобочитаемого и обслуживаемого кода. Как и в случае с Lego, эти шаблоны можно использовать в качестве строительных блоков для создания надежного программного проекта.

Apache Camel также поддерживает широкий спектр коннекторов для интеграции вашего приложения с различными платформами и технологиями. И, кстати, он также прекрасно сочетается с Spring .

Если вы не знакомы с Spring, вам может пригодиться этот пост:  Обработка ленты Twitter с использованием Spring Boot .

В следующих разделах мы рассмотрим пример приложения, в котором Camel интегрирован как с примером канала Twitter, так и с ElasticSearch.

Что такое ElasticSearch?

ElasticSearch, похожий на Apache Solr, представляет собой высоко масштабируемую систему полнотекстового поиска с открытым исходным кодом на основе Java, построенную на основе Apache Lucene .

В этом примере приложения мы собираемся использовать ElasticSearch для индексирования твитов в режиме реального времени, а также для обеспечения полнотекстового поиска по этим твитам.

Другие используемые технологии

Помимо Apache Camel и ElasticSearch, я также включил в это приложение и другие фреймворки:  Gradle  в качестве инструмента сборки,  Spring Boot в качестве фреймворка веб-приложения и Twitter4j  для чтения твитов из примера фида Twitter.

Начиная

Скелет проекта был создан по адресу  http://start.spring.io,  где я проверил параметр веб-зависимости, заполнил раздел «Метаданные проекта» и выбрал «Gradle Project» в качестве типа проекта.

После того, как проект создан, вы можете загрузить и импортировать его в свою любимую среду IDE. Сейчас я не буду вдаваться в подробности о Gradle, но вот список всех зависимостей в файле build.gradle :

def camelVersion = '2.15.2'
dependencies {
    compile("org.springframework.boot:spring-boot-starter-web")
    compile("org.apache.camel:camel-core:${camelVersion}")
    compile("org.apache.camel:camel-spring-boot:${camelVersion}")
    compile("org.apache.camel:camel-twitter:${camelVersion}")
    compile("org.apache.camel:camel-elasticsearch:${camelVersion}")
    compile("org.apache.camel:camel-jackson:${camelVersion}")
    compile("joda-time:joda-time:2.8.2")
    testCompile("org.springframework.boot:spring-boot-starter-test")
}

from("file://orders").
  convertBodyTo(String.class).
  to("log:com.mycompany.order?level=DEBUG").
  to("jms:topic:OrdersTopic");

from("file://orders").
  convertBodyTo(String.class).
  to("direct:orders");

from("direct:orders).
  to("log:com.mycompany.order?level=DEBUG").
  to("jms:topic:OrdersTopic");

    private String ES_TWEET_INDEXER_ENDPOINT = "direct:tweet-indexer-ES";

...

        from("twitter://streaming/sample?type=EVENT&consumerKey={{twitter4j.oauth.consumerKey}}&consumerSecret={{twitter4j.oauth.consumerSecret}}&accessToken={{twitter4j.oauth.accessToken}}&accessTokenSecret={{twitter4j.oauth.accessTokenSecret}}")
            .to(ES_TWEET_INDEXER_ENDPOINT)
        ;

    @Value("${elasticsearch.tweet.uri}")
    private String elasticsearchTweetUri;

...

        from(ES_TWEET_INDEXER_ENDPOINT)
            // groups tweets into separate indexes on a weekly basis to make it easier clean up old tweets:
            .process(new WeeklyIndexNameHeaderUpdater(ES_TWEET_INDEX_TYPE))
            // converts Twitter4j Tweet object into an elasticsearch document represented by a Map:
            .process(new ElasticSearchTweetConverter())
            // collects tweets into weekly batches based on index name:
            .aggregate(header("indexName"), new ListAggregationStrategy())
                // creates new batches every 2 seconds
                .completionInterval(2000)
                // makes sure the last batch will be processed before the application shuts down:
                .forceCompletionOnStop()
            // inserts a batch of tweets to elasticsearch: 
            .to(elasticsearchTweetUri)
            .log("Uploaded documents to ElasticSearch index ${headers.indexName}: ${body.size()}")
        ;

    public static final String TWEET_SEARCH_URI = "vm:tweetSearch";

...

        from(TWEET_SEARCH_URI)
            .setHeader("CamelFileName", simple("tweet-${body}-${header.maxSize}-${date:now:yyyyMMddHHmmss}.txt"))
            // calls the search() method of the esTweetService which returns an iterator
            // to process search result - better than keeping the whole resultset in memory:
            .split(method(esTweetService, "search"))
                // converts Elasticsearch doucment to Map object:
                .process(new ElasticSearchSearchHitConverter())
                // serializes the Map object to JSON:
                .marshal(new JacksonDataFormat())
                // appends new line at the end of every tweet
                .setBody(simple("${body}\n"))
                // write search results as json into a file under /tmp folder:
                .to("file:/tmp?fileExist=Append")
            .end()
            .log("Wrote search results to /tmp/${headers.CamelFileName}")
        ;

    @Autowired
    private ProducerTemplate producerTemplate;

    @RequestMapping(value = "/tweet/search", method = { RequestMethod.GET, RequestMethod.POST },
            produces = MediaType.TEXT_PLAIN_VALUE)
    @ResponseBody
    public String tweetSearch(@RequestParam("q") String query,
                              @RequestParam(value = "max") int maxSize) {
        LOG.info("Tweet search request received with query: {} and max: {}", query, maxSize);
        Map<String, Object> headers = new HashMap<String, Object>();
        // "content" is the field in the Elasticsearch index that we'll be querying:
        headers.put("queryField", "content");
        headers.put("maxSize", maxSize);
        producerTemplate.asyncRequestBodyAndHeaders(CamelRouter.TWEET_SEARCH_URI, query, headers);
        return "Request is queued";
    }

    public SearchHitIterator search(@Body String query, @Header(value = "queryField") String queryField, @Header(value = "maxSize") int maxSize) {
        boolean scroll = maxSize > batchSize;
        LOG.info("Executing {} on index type: '{}' with query: '{}' and max: {}", scroll ? "scan & scroll" : "search", indexType, query, maxSize);
        QueryBuilder qb = termQuery(queryField, query);

        long startTime = System.currentTimeMillis();
        SearchResponse response = scroll ? prepareSearchForScroll(maxSize, qb) : prepareSearchForRegular(maxSize, qb);
        return new SearchHitIterator(client, response, scroll, maxSize, KEEP_ALIVE_MILLIS, startTime);
    }

package com.kaviddiss.twittercamel;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

$GRADLE_HOME/bin/gradlew build && java -jar build/libs/twitter-camel-ingester-0.0.1-SNAPSHOT.jar