Разработка сервиса агрегации открытых данных

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

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

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

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

Список используемых источников

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