ОБОБЩЕНИЕ РЕЗУЛЬТАТОВ И ПРАКТИЧЕСКИЕ РЕКОМЕНДАЦИИ

Помимо выполнения всех поставленных задач, попутно были решены также и задачи, которые не были поставлены изначально. К ним относятся такие результаты, которые вытекают из выбора гибких и современных технологий, которым сейчас стараются придерживаться многие передовые корпорации. Речь идет о кроссплатформенной разработке, а именно о языке программирования Java. За счет использования этого языка в создании сервиса, удалось достичь независимости от платформы, на котором необходимо развертывать приложения. Благодаря этому, становится возможным запускать сервис на любой операционной системе, где была предустановлена виртуальная машина JRE (Java Runtime Environment). Помимо этого, микросервисная архитектура позволяет выделять отдельные независимые программные компоненты. Из этого следует, что каждый программный модуль может быть запущен более чем в одном экземпляре. Таким образом, становится возможным горизонтальное масштабирование системы. Другими словами, с ростом количества посетителей на сайт или с ростом количества сообщений из социальных сетей можно почти линейно увеличивать вычислительные мощности. Более того, из-за того, что в системе выделены отдельные модули, становится простым процесс их контейнеризации. В качестве этой технологии можно выбрать самый популярный фреймворк для создания и запуска контейнером - Docker. Все это сделало данных сервис более гибким к развертыванию и устойчивым к нагрузкам, что является важной деталью для систем мониторинга в реальном времени.

На данный момент в сервисе реализованы клиенты для двух социальных сетей: Twitter и Instagram. Это одни из самых популярных на сегодняшний день платформ в мире. Они имеют развитый и технически сложный программный интерфейс, позволяющие выстраивать гибкие запросы к своим базам данных. Интеграция с другими сетями была осложнена рядом факторов, которые ранее уже были описаны в данной работе. Тем не менее, ограничение клиентов было вызвано только ограничением по времени написания сервиса. Стоит отметить, что количество клиентов под разные социальные сети является хорошей точкой роста и развития для сервиса. Дело в том, что чем больше сообщений будет обрабатывать и сервис агрегации, тем точнее должна получаться визуализация активности пользователей. В последствие можно будет добавить другие социальные сети, например, мирового лидера в лице Facebook и более успешного игрока на российском рынке - Вконтакте.

Использование современных технологий разработки оставляет возможность для расширения функционала. Есть масса идей, которые позволят качественно улучшить сервис и развить в нем свой потенциал. Каждый день или любой другой отрезок времени отслеживать, где пользователи наиболее активно себя проявляют в социальных сетях. Далее с помощью загруженных наборов открытых данных можно сопоставлять их географическое положение и определять, какой объект является самым популярным на это время. Это может послужить своего рода рекомендательной системой для отслеживания трендов посещений пользователями различных заведений, музеев или выставок.

НОВИЗНА И ПРАКТИЧЕСКАЯ ЗНАЧИМОСТЬ ИССЛЕДОВАНИЯ

Отличительной особенностью среди всех сервисов на основе открытых данных является тот факт, что визуализация информации является динамической. Это достигается за счет агрегации потоков данных из социальных сетей в реальном времени. Недостатком сервисов, которые основываются только лишь на открытых данных, является их редкое обновление на порталах. Как правило, такие наборы данных статические и модифицируются крайне редко. Это влечет за собой такие последствия, как проблемы с удержанием интернет аудитории у веб-ресурса. Большинство пользователей, которые заходят на подобные сайты, редко возвращаются или пользуются им повторно. Это объясняется тем, что сервис был спроектирован таким образом, чтобы обратить внимание на какую-либо проблему или явление. После того, как люди, пришедшие на интернет портал, изучили поднятый вопрос, необходимость в нем сразу отпадает. Примером может послужить проект, участвовавший в хакатоне по открытым данным. Проект заключается в том, чтобы показать общественности и властям, что некоторые велодорожки были проложены не там, где движутся велосипедисты в реальной жизни. Такие проекты можно считать одноразовыми, поскольку они полностью статичны. После посещения такого сайта интерес к данной теме теряется, а в случае повторного посещения ресурса пользователь не обнаружит ничего нового.

Иначе дело обстоит с сервисом агрегации данных. Дело в том, что сама природа социальных сетей автоматически избавляет данный сервис от подобных недостатков. Благодаря тому, что все наполнение сетей целиком и полностью генерируется пользователями, такая экосистема сразу становится динамической. Соответственно, все сервисы, которые используют пользовательскую активность в своей работе, также становятся динамическими. Все это ведет к тому, что в каждый момент времени состояние сервиса уникально. Именно это и дает преимущество перед статическими веб ресурсами, ограниченных определенным набором открытых данных. Дело в том, что с точки зрения пользователей гораздо выше вероятность повторного посещения подобного рода сайтов, поскольку они не знают наверняка, что там увидят. Изучение активности пользователей в привязке к наборам открытых данных теперь может быть растянуто по времени. Другими словами, можно выявлять значимые тренды поведения пользователей в социальных сетях, определять связь между местами скопления пользователей и объектами города, такими как парки, скверы, места общественного питания и прочие. На момент написания работы разработанный сервис не имеет аналогов.



ЗАКЛЮЧЕНИЕ

В результате выполнения выпускной квалификационной работы был реализован сервис агрегации открытых данных и данных из социальных сетей. При этом была достигнута основная цель данной работы. Стоит отметить, что автором были решены все поставленные задачи. Данная система состоит из ряда отдельных независимых модулей, выполненных согласно микросервисной архитектуре. Это клиенты под каждую поддерживаемую социальную сеть, адаптеры для конвертации сообщений в единый универсальный формат, сервис базы данных и веб-сервер для предоставления доступа к данным пользователю. Сервис представляет собой веб-сайт, который можно использовать с помощью браузера. Разработанный программный комплекс, сочетающий в себе средства для визуализации и агрегации открытых данных и данных из социальных сетей, не имеет аналогов на момент написания данной работы.

Главной особенностью данного проекта является тот факт, что с помощью сервиса можно выявлять зависимости между географическим положением пользовательской активности, ее интенсивностью и объектами из набора открытых данных. Так, благодаря интуитивно понятной визуализации вышеперечисленных данных, можно анализировать места скопления людей, которые являются пользователями социальных сетей. С помощью данного сервиса можно выявлять закономерности в расположении объектов города, например парков и музеев, и пользовательской активностью. Благодаря ему можно понять, какие места пользуются спросом у людей, а какие являются непопулярными. Например, каким паркам, помимо ухода за территорией, стоит обратить внимание на наполнение запоминающимися объектами. К ним можно отнести арт-объекты или памятники. Все это будет вести к повышению вовлеченности посетителей, что в свою очередь приведет к их желанию поделиться своими впечатлениями со своими друзьями и знакомыми в социальных сетях. Более того, данный сервис поможет выявить такое явление как сезонность. Такая информация может быть полезна во многих сферах жизни, например, представителям рекламных агентств или сетей общественного питания.

Также следует обратить внимание, что в связи с использованием самых современных инструментов и технологий индустрии разработки программного обеспечения, разработанный сервис получился очень гибким в плане поддержки и развертывания. Дело в том, что система является независимой от платформы. Другими словами, она может выполняться на любой операционной системе, где есть виртуальная машина Java. Также в виду использования микросервисной архитектуры, отдельные независимые сервисы легко поддаются контейнеризации. Такой подход сейчас стараются поддерживать все передовые корпорации, занимающиеся разработкой программного обеспечения и облачными технологиями. Самым распространенным способом является использование технологии Docker, позволяющей без проблем производить горизонтальное масштабирование приложений и удобное администрирование систем.

Несмотря на то, что сервис представляет собой готовый к использованию программный продукт, существует несколько способов для его улучшения. В первую очередь, в систему может быть добавлена автоматическая рекомендательная система, которая способна распознавать тренды активности пользователей в социальных сетях, сопоставлять с наборами открытых данных и делать выводы о том, какие места в данный момент пользуются популярностью. Также стоит отметить, что с помощью добавления в систему новых клиентов для других социальны сетей, точность работы сервиса должна увеличиться. Это можно объяснить тем, что такая тепловая карта будет еще более показательная, а визуализация даст наиболее полную картину. Таким образом, у проекта существует потенциал для развития и роста.



СПИСОК ИСПОЛЬЗУЕМЫХ ИСТОЧНИКОВ

1. Boyd D., Ellison N.B. Social Network sites: Definition, history and scholarship // Journal of Computer-Mediated Communication, 2011, №11.

2. Chan Y. “A Distributed Stream Library for Java 8,” Ph.D. dissertation, University of York, 2016.

3. Chowdhury M., Zaharia M., Stoica I., Performance and Scalability of Broadcast in Spark. 2010.

4. Dewire D.T. Client-server computing. McGrawHill, Singapore, 1993.

5. Digital Fuel of the 21st Century: Innovation through Open Data and the Network Effect, Harvard University, 2011.

6. Eur-Lex, Open data An engine for innovation, growth and transparent governance, 2011.

7. Google Maps Documentation [Электронный ресурс] // URL:https://developers.google.com/maps/documentation (Дата обращения 02.03.2017).

8. Grosso W. Java RMI. First Edition. O'Reilly and Associates, USA, 2001.

9. Kumar K., Liu J., Lu Y., Bhargava B., “A survey of computation offloading for mobile systems,” Mobile Networks and Applications, vol. 18, no. 1, pp. 129-140, 2013.

10. Lahiri M., Berger-Wolf T., “Mining Periodic Behavior in Dynamic Social Networks,” Proceedings of the 8th IEEE Inter-national Conference on Data Mining, 2008, pp. 373-382.

11. Microservices [Электронный ресурс] // URL: http://microservices.io/patterns/microservices.html (Дата обращения 20.04.2017).

12. O'Reilly T., What Is Web 2.0 [Электронный ресурс] // URL: http://www.oreillynet.com/pub/a/oreilly/tim/news/2005/09/30/what-is-web-20.html (Дата обращения 16.02.2017).

13. Open Data Incubator Europe [Электронный ресурс] // URL: https://opendataincubator.eu/ (Дата обращения 16.02.2017).

14. OpenData 500 [Электронный ресурс] // URL: http://www.opendata500.com/ (Дата обращения 16.02.2017).

15. OpenStreetMap [Электронный ресурс] // URL: https://www.openstreetmap.org/ (Дата обращения 02.03.2017).

16. Ricardo T., Marco T.V., Roberto S.B. An approach for extracting modules from monolithic software architectures. Workshop, pages 1-18, 2012.

17. Scharl, A. Towards the Geospatial Web: Media Platforms for Managing Geotagged Knowledge Repositories. The Geospatial Web Geobrowsers, Social Software and the Web 2.0 are Shaping the Network Society (pp. 3-14). London: Springer, 2007.

18. Twitter Developer Documentation [Электронный ресурс] // URL: https://dev.twitter.com/overview/api/tweets (Дата обращения 18.04.2017).

19. Venner J., Pro Hadoop. Apress, June 22, 2009.

20. Wang G., Xiong Y., Yun J., Cavallaro J. “Accelerating computer vision algorithms using OpenCL framework on the mobile GPU--a case study,” in Proceedings of the 38th IEEE International Conference on Acoustics, Speech, and Signal Processing (ICASSP '13), IEEE, Vancouver, Canada, May 2013.

21. Бегтин И.В. Проблема открытых данных в России // Земля из космоса. 2011. №11. С.20-25.

22. Борисенко О.Д., Турдаков Д. Ю., Кузнецов С. Д., Автоматическое создание виртуальных кластеров Apache Spark в облачной среде OpenStack. Труды Института системного программирования РАН, том 17, 2009 г. Стр 31-50.

23. Катков Е.В., Сорочайкин А.Н. Моделирование процессов инновационного развития предприятий // Вестник Самарского государственного университета. 2012. №10. С.33-39.

24. Остервальдер А., Пинье И. Построение бизнес-моделей. Настольная книга стратега и новатора. -- М.: Альпина Паблишер, 2011.

25. Открытые государственные финансовые данные [Электронный ресурс] // URL: http://budgetapps.ru (Дата обращения 16.02.2017).

26. Парамонов В., ГОС: Объем информации в интернете удваивается каждые полтора года [Электронный ресурс] // URL: http://www.newsland.ru/news/detail/id/367158/ (Дата обращения 15.02.2017).

27. Портал открытых данных правительства Москвы [Электронный ресурс] // URL: https://data.mos.ru/ (Дата обращения 18.04.2017).

28. Портал открытых данных Российской Федерации [Электронный ресурс] // URL: http://data.gov.ru/ (Дата обращения 18.04.2017).

29. Хакатон портала открытых данных data.gov.ru [Электронный ресурс] // URL: http://data.gov.ru/hackathon (Дата обращения 16.02.2017).

30. Число пользователей Facebook превысило 1,71 миллиарда [Электронный ресурс] // URL: https://rg.ru/2016/07/28/chislo-polzovatelej-facebook-prevysilo-171-milliarda.html (Дата обращения 16.02.2017).

31. Что такое открытые данные? [Электронный ресурс] // URL: https://opengovdata.ru/definition/ (Дата обращения 16.02.2017).

32. Яндекс. API карт. Документация. [Электронный ресурс] // URL: https://tech.yandex.ru/maps/doc/jsapi/2.1/quick-start/tasks/quick-start-docpage/ (Дата обращения 02.03.2017).



ПРИЛОЖЕНИЕ. ТЕКСТ ПРОГРАММЫ

MapController.java

package com.alexcodes.web.controller;

import com.alexcodes.web.dto.MapDTO;

import com.alexcodes.web.service.MapService;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.http.MediaType;

import org.springframework.util.Assert;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RestController;

@RestController

@RequestMapping("/map")

public class MapController {

private final MapService mapService;

@Autowired

public MapController(MapService mapService) {

Assert.notNull(mapService, "Cannot be null");

this.mapService = mapService;

}

@RequestMapping(value = "/coordinates",

method = RequestMethod.GET,

produces = MediaType.APPLICATION_JSON_VALUE)

public MapDTO getCoordinates() {

return mapService.getMap();

}

}

CoordinatesService.java

package com.alexcodes.web.service;

import java.util.List;

public interface CoordinatesService {

List<List<Double>> findCoordinates();

}

MapService.java

package com.alexcodes.web.service;

import com.alexcodes.web.dto.MapDTO;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import org.springframework.util.Assert;

@Service

public class MapService {

private final CoordinatesService coordinatesService;

@Autowired

public MapService(CoordinatesService coordinatesService) {

Assert.notNull(coordinatesService, "Cannot be null");

this.coordinatesService = coordinatesService;

}

public MapDTO getMap() {

MapDTO dto = new MapDTO();

dto.coordinates = coordinatesService.findCoordinates();

return dto;

}

}

SimpleCoordinatesService.java

package com.alexcodes.web.service;

import com.alexcodes.common.dao.GeoPostRepository;

import com.alexcodes.common.domain.GeoPost;

import com.google.common.collect.Lists;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Profile;

import org.springframework.stereotype.Service;

import java.util.Arrays;

import java.util.List;

import java.util.Objects;

import java.util.stream.Collectors;

@Service

@Profile("default")

public class SimpleCoordinatesService implements CoordinatesService {

private final GeoPostRepository geoPostRepository;

@Autowired

public SimpleCoordinatesService(GeoPostRepository geoPostRepository) {

this.geoPostRepository = geoPostRepository;

}

@Override

public List<List<Double>> findCoordinates() {

List<GeoPost> posts = Lists.newArrayList(geoPostRepository.findAll());

return posts.stream()

.map(post -> post.location)

.filter(Objects::nonNull)

.filter(location -> location.latitude != 55.7547875 && location.longitude != 37.427642500000005)

.map(location -> Arrays.asList(location.latitude, location.longitude))

.collect(Collectors.toList());

}

}

CoordinateDTO.java

package com.alexcodes.web.dto;

public class CoordinateDTO {

public double longitude;

public double latitude;

public CoordinateDTO() {

}

public CoordinateDTO(double longitude, double latitude) {

this.longitude = longitude;

this.latitude = latitude;

}

}

MapDTO.java

package com.alexcodes.web.dto;

import java.time.Instant;

import java.util.List;

public class MapDTO {

public Instant lastModified;

public List<List<Double>> coordinates;

}

WebMain.java

package com.alexcodes.web;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.ComponentScan;

@SpringBootApplication

@ComponentScan(basePackages = "com.alexcodes.*")

public class WebMain {

public static void main(String[] args) {

SpringApplication.run(WebMain.class);

}

}

Index.html

<!DOCTYPE html>

<html lang="en">

<head>

<meta charset="utf-8">

<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">

<title>Social Parks</title>

<link href="https://yandex.st/bootstrap/2.3.2/css/bootstrap.min.css" rel="stylesheet">

<style type="text/css">

html, body, .hero-unit {

min-height: 100%;

height: 100%;

margin: 0;

}

#YMapsID {

width: 900px;

height: 700px;

}

#YMapsCode {

width: 880px;

}

</style>

<script src="js/jquery-3.2.1.min.js" type="text/javascript"></script>

<script src="http://api-maps.yandex.ru/2.1/?lang=ru_RU" type="text/javascript"></script>

<script src="js/heatmap.min.js" type="text/javascript"></script>

<script type="text/javascript">

ymaps.ready(function () {

$.get("/map/coordinates", function (response) {

var data = response.coordinates;

var map = new ymaps.Map('YMapsID', {

center: [55.751588, 37.617861],

controls: ['zoomControl', 'typeSelector', 'fullscreenControl'],

zoom: 11, type: 'yandex#satellite'

}),

buttons = {

dissipating: new ymaps.control.Button({

data: {

content: 'Toggle dissipating'

},

options: {

selectOnClick: false,

maxWidth: 150

}

}),

opacity: new ymaps.control.Button({

data: {

content: 'Change opacity'

},

options: {

selectOnClick: false,

maxWidth: 150

}

}),

radius: new ymaps.control.Button({

data: {

content: 'Change radius'

},

options: {

selectOnClick: false,

maxWidth: 150

}

}),

gradient: new ymaps.control.Button({

data: {

content: 'Reverse gradient'

},

options: {

selectOnClick: false,

maxWidth: 150

}

}),

heatmap: new ymaps.control.Button({

data: {

content: 'Toggle Heatmap'

},

options: {

selectOnClick: false,

maxWidth: 150

}

})

},

gradients = [{

0.1: 'rgba(128, 255, 0, 0.7)',

0.2: 'rgba(255, 255, 0, 0.8)',

0.7: 'rgba(234, 72, 58, 0.9)',

1.0: 'rgba(162, 36, 25, 1)'

}, {

0.1: 'rgba(162, 36, 25, 0.7)',

0.2: 'rgba(234, 72, 58, 0.8)',

0.7: 'rgba(255, 255, 0, 0.9)',

1.0: 'rgba(128, 255, 0, 1)'

}],

radiuses = [5, 10, 20, 30],

opacities = [0.4, 0.6, 0.8, 1];

ymaps.modules.require(['Heatmap'], function (Heatmap) {

var heatmap = new Heatmap(data, {

gradient: gradients[0],

radius: radiuses[1],

opacity: opacities[2]

});

heatmap.setMap(map);

buttons.dissipating.events.add('press', function () {

heatmap.options.set(

'dissipating', !heatmap.options.get('dissipating')

);

});

buttons.opacity.events.add('press', function () {

var current = heatmap.options.get('opacity'),

index = opacities.indexOf(current);

heatmap.options.set(

'opacity', opacities[++index == opacities.length ? 0 : index]

);

});

buttons.radius.events.add('press', function () {

var current = heatmap.options.get('radius'),

index = radiuses.indexOf(current);

heatmap.options.set(

'radius', radiuses[++index == radiuses.length ? 0 : index]

);

});

buttons.gradient.events.add('press', function () {

var current = heatmap.options.get('gradient');

heatmap.options.set(

'gradient', current == gradients[0] ? gradients[1] : gradients[0]

);

});

buttons.heatmap.events.add('press', function () {

heatmap.setMap(

heatmap.getMap() ? null : map

);

});

for (var key in buttons) {

if (buttons.hasOwnProperty(key)) {

map.controls.add(buttons[key]);

}

}

});

});

});

</script>

</head>

<body>

<div class="hero-unit">

<div class="container">

<p>Yandex Maps API <a href="https://github.com/yandex/mapsapi-heatmap">Heatmap Module</a></p>

<div id="YMapsID"></div>

</div>

</div>



</body>

</html>

MongoConfiguration.java

package com.alexcodes.common.config;

import com.mongodb.Mongo;

import com.mongodb.MongoClient;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Configuration;

import org.springframework.data.mongodb.config.AbstractMongoConfiguration;

import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;

@Configuration

@EnableMongoRepositories({"com.alexcodes.common.dao"})

public class MongoConfiguration extends AbstractMongoConfiguration {

@Value("${spring.data.mongodb.host}")

private String host;

@Value("${spring.data.mongodb.port}")

private int port;

@Value("${spring.data.mongodb.database}")

private String database;

@Override

protected String getDatabaseName() {

return database;

}

@Override

public Mongo mongo() throws Exception {

return new MongoClient(host, port);

}

}

GeoPost.java

package com.alexcodes.common.domain;

import com.google.common.base.MoreObjects;

import org.springframework.data.annotation.Id;

import java.time.Instant;

public class GeoPost {

@Id

public String id;

public SourceType sourceType;

public String text;

public Location location;

public Instant timestamp;

@Override

public String toString() {

return MoreObjects.toStringHelper(this)

.add("sourceType", sourceType)

.add("text", text)

.add("location", location)

.add("timestamp", timestamp)

.toString();

}

}

Location.java

package com.alexcodes.common.domain;

import com.google.common.base.MoreObjects;

public class Location {

public Type type;

public double longitude;

public double latitude;

public Location() {}

public Location(Type type, double longitude, double latitude) {

this.type = type;

this.longitude = longitude;

this.latitude = latitude;

}

@Override

public String toString() {

return MoreObjects.toStringHelper(this)

.add("longitude", longitude)

.add("latitude", latitude)

.toString();

}

public enum Type {

POINT, CITY

}

}

AppConfig.java

package com.alexcodes.twitter.config;

import com.twitter.hbc.core.endpoint.Location;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class TwitterConfig {

@Value("${twitter.location.southwest.longitude}")

private double southWestLongitude;

@Value("${twitter.location.southwest.latitude}")

private double southWestLatitude;

@Value("${twitter.location.northeast.longitude}")

private double northEastLongitude;

@Value("${twitter.location.northeast.latitude}")

private double northEastLatitude;

@Bean

public Location location() {

return new Location(

new Location.Coordinate(southWestLongitude, southWestLatitude),

new Location.Coordinate(northEastLongitude, northEastLatitude));

}

}

LocationExtractor.java

package com.alexcodes.twitter.logic;

import com.alexcodes.common.domain.Location;

import com.google.gson.JsonArray;

import com.google.gson.JsonElement;

import com.google.gson.JsonObject;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;

import java.util.ArrayList;

import java.util.List;



@Service

public class LocationExtractor {

private static final Logger log = LoggerFactory.getLogger(LocationExtractor.class);

private static final String COORDINATES = "coordinates";

private static final String PLACE = "place";

private static final String PLACE_TYPE = "place_type";

private static final String CITY = "city";

private static final String BOUNDING_BOX = "bounding_box";

private static final String TYPE = "type";

private static final String POLYGON = "Polygon";

public Location getLocation(JsonObject root) {

JsonElement coordinates = root.get(COORDINATES);

if (coordinates != null && !coordinates.isJsonNull())

return extractCoordinates(coordinates);

JsonElement place = root.get(PLACE);

if (place != null && !place.isJsonNull())

return extractPlace(place);

log.error("Cannot extract location from {}", root);

return null;

}

private Location extractCoordinates(JsonElement coordinates) {

JsonArray array = coordinates.getAsJsonObject().getAsJsonArray(COORDINATES);

double longitude = array.get(0).getAsDouble();

double latitude = array.get(1).getAsDouble();

return new Location(Location.Type.POINT, longitude, latitude);

}

private Location extractPlace(JsonElement place) {

String placeType = place.getAsJsonObject().get(PLACE_TYPE).getAsString();

JsonObject boundingBox = place.getAsJsonObject().getAsJsonObject(BOUNDING_BOX);

if (!placeType.equals(CITY)) {

log.warn("Unknown placeType: {}", placeType);

return null;

}

Point point = extractBoundingBox(boundingBox);

return new Location(Location.Type.CITY, point.longitude, point.latitude);

}

private Point extractBoundingBox(JsonObject boundingBox) {

String type = boundingBox.getAsJsonPrimitive(TYPE).getAsString();

switch (type) {

case POLYGON: {

JsonArray coordinates = boundingBox.getAsJsonArray(COORDINATES)

.get(0)

.getAsJsonArray();

List<Double> longs = new ArrayList<>(coordinates.size());

List<Double> lats = new ArrayList<>(coordinates.size());

for (int i = 0; i < coordinates.size(); i++) {

JsonArray point = coordinates.get(i).getAsJsonArray();

longs.add(point.get(0).getAsDouble());

lats.add(point.get(1).getAsDouble());

}

double avgLong = longs.stream()

.mapToDouble(Double::doubleValue)

.average()

.getAsDouble();

double avgLat = lats.stream()

.mapToDouble(Double::doubleValue)

.average()

.getAsDouble();

return new Point(avgLong, avgLat);

}

default:

log.error("Unknown format: {}", boundingBox);

return new Point(0.0, 0.0);

}

}

private static class Point {

public double longitude;

public double latitude;

public Point(double longitude, double latitude) {

this.longitude = longitude;

this.latitude = latitude;

}

}

}

PostConverter.java

package com.alexcodes.twitter.logic;

import com.alexcodes.common.domain.GeoPost;

import com.alexcodes.common.domain.SourceType;

import com.alexcodes.common.logic.PostConverter;

import com.google.gson.JsonElement;

import com.google.gson.JsonObject;

import com.google.gson.JsonParser;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

import java.time.Instant;

import java.time.format.DateTimeFormatter;

@Service

public class TweetConverter implements PostConverter {

private static final String TEXT = "text";

private static final String CREATED_AT = "created_at";

private static final DateTimeFormatter formatter =

DateTimeFormatter.ofPattern("E MMM dd HH:mm:ss Z yyyy");

private final LocationExtractor locationExtractor;

@Autowired

public TweetConverter(LocationExtractor locationExtractor) {

this.locationExtractor = locationExtractor;

}

@Override

public GeoPost convert(String message) {

JsonParser parser = new JsonParser();

JsonObject root = parser.parse(message).getAsJsonObject();

GeoPost post = new GeoPost();

post.sourceType = SourceType.TWITTER;

post.text = getText(root);

post.location = locationExtractor.getLocation(root);

post.timestamp = getTimestamp(root);

return post;

}

private String getText(JsonObject root) {

return root.getAsJsonPrimitive(TEXT).getAsString();

}

private Instant getTimestamp(JsonObject root) {

JsonElement createdAt = root.get(CREATED_AT);

if (createdAt == null) return null;

String timestamp = createdAt.getAsString();

return formatter.parse(timestamp).query(Instant::from);

}

}



PostProcessor.java

package com.alexcodes.twitter.logic;

import com.alexcodes.common.dao.GeoPostRepository;

import com.alexcodes.common.domain.GeoPost;

import com.alexcodes.common.logic.PostProcessor;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Service;

@Service

public class TweetProcessor implements PostProcessor {

private static final Logger log = LoggerFactory.getLogger(TweetProcessor.class);

private final TweetConverter tweetConverter;

private final GeoPostRepository geoPostRepository;

@Autowired

public TweetProcessor(

TweetConverter tweetConverter,

GeoPostRepository geoPostRepository) {

this.tweetConverter = tweetConverter;

this.geoPostRepository = geoPostRepository;

}

@Override

public void process(String message) {

try {

GeoPost post = tweetConverter.convert(message);

log.debug("Tweet: {}", post);

geoPostRepository.save(post);

} catch (RuntimeException e) {

log.error("Exception during tweet processing:", e);

}

}

}

Listener.java

package com.alexcodes.twitter.service;

import com.alexcodes.twitter.logic.TweetProcessor;

import com.twitter.hbc.ClientBuilder;

import com.twitter.hbc.core.Client;

import com.twitter.hbc.core.Constants;

import com.twitter.hbc.core.Hosts;

import com.twitter.hbc.core.HttpHosts;

import com.twitter.hbc.core.endpoint.Location;

import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint;

import com.twitter.hbc.core.event.Event;

import com.twitter.hbc.core.processor.StringDelimitedProcessor;

import com.twitter.hbc.httpclient.auth.Authentication;

import com.twitter.hbc.httpclient.auth.OAuth1;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.boot.CommandLineRunner;

import org.springframework.stereotype.Service;

import org.springframework.util.Assert;

import java.util.Collections;

import java.util.List;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

@Service

public class TwitterListener implements CommandLineRunner {

private static final Logger log = LoggerFactory.getLogger(TwitterListener.class);

private static final int QUEUE_SIZE = 1000;

@Value("${twitter.consumerKey}")

private String consumerKey;

@Value("${twitter.consumerSecret}")

private String consumerSecret;

@Value("${twitter.token}")

private String token;

@Value("${twitter.tokenSecret}")

private String tokenSecret;

@Value("${twitter.client.name}")

private String clientName;

private final Location location;

private final TweetProcessor tweetProcessor;

@Autowired

public TwitterListener(Location location, TweetProcessor tweetProcessor) {

Assert.notNull(location, "Location cannot be null");

Assert.notNull(tweetProcessor, "Cannot be null");

this.location = location;

this.tweetProcessor = tweetProcessor;

}

public void run(String... strings) throws Exception {

// Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream

BlockingQueue<String> msgQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);

BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);

// Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth)

Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);

StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();

// Optional: set up some followings and track terms

List<Location> locations = Collections.singletonList(location);

hosebirdEndpoint.locations(locations);

// These secrets should be read from a config file

Authentication hosebirdAuth = new OAuth1(consumerKey, consumerSecret, token, tokenSecret);

ClientBuilder builder = new ClientBuilder()

.name(clientName)

.hosts(hosebirdHosts)

.authentication(hosebirdAuth)

.endpoint(hosebirdEndpoint)

.processor(new StringDelimitedProcessor(msgQueue))

.eventMessageQueue(eventQueue);

Client hosebirdClient = builder.build();

hosebirdClient.connect();

int i = 0;

long time = System.currentTimeMillis();

// on a different thread, or multiple different threads....

while (!hosebirdClient.isDone()) {

String message = msgQueue.take();

// log.debug("Receive: {}", message);

tweetProcessor.process(message);

i++;

// if (System.currentTimeMillis() - time > 60_000L) break;

}

log.info("{} TPS", i);

hosebirdClient.stop();

}

}

Размещено на Allbest.ru

...