Введение
Для анализа потоковых данных необходимы источники этих данных.
Так же важна сама информация, которая предоставляется источниками.
А источники с текстовой информацией, к примеру, еще и редки.
Из интересных источников можно выделить следующие: twitter, vk. Но эти источники подходят не под все задачи.
Есть источники с нужными данными, но эти источники не потоковые.
Здесь можно привести следующее ссылки: public-apis.
При решении задач, связанных с потоковыми данными, можно
воспользоваться старым способом.
Скачать данные и отправить в поток.
Для примера можно воспользоваться следующим источником: imdb.
Следует отметить, что imdb предоставляет данные самостоятельно. См.
IMDb Datasets. Но можно принять, что данные
собранные напрямую содержат более актуальную информацию.
Язык: Java 1.8.
Библиотеки: kafka 2.6.0, jsoup 1.13.1.
Сбор данных
Сбор данных представляет из себя сервис, который по входным
данным загружает html-страницы, ищет нужную информацию и
преобразует в набор объектов.
Итак источник данных: imdb. Информация будет собираться о фильмах и будет
использован следующий запрос: https://www.imdb.com/search/title/?release_date=%s,%s&countries=%s
Где 1, 2 параметр это даты. 3 параметр страны.
Для лучшего понимания источника данных можно обратится к следующему
ресурсу: imdb-extensive-dataset.
Интерфейс для сервиса:
public interface MovieDirectScrapingService { Collection<Movie> scrap();}
Класс Movie это класс, которые содержит информацию об одном фильме (или о шоу и т.п.).
class Movie { public final String titleId; public final String titleUrl; public final String title; public final String description; public final Double rating; public final String genres; public final String runtime; public final String baseUrl; public final String baseNameUrl; public final String baseTitleUrl; public final String participantIds; public final String participantNames; public final String directorIds; public final String directorNames;
Анализ данных на одной странице.
Информация собирается следующим образом. Данные закачиваются с
помощью jsoup. Далее ищутся нужные html-элементы и трансформируются
в экземпляры для фильмов.
String scrap(String url, List<Movie> items) { Document doc = null; try { doc = Jsoup.connect(url).header("Accept-Language", language).get(); } catch (IOException e) { e.printStackTrace(); } if (doc != null) { collectItems(doc, items); return nextUrl(doc); } return "";}
Поиск ссылки на следующею страницу.
String nextUrl(Document doc) { Elements nextPageElements = doc.select(".next-page"); if (nextPageElements.size() > 0) { Element hrefElement = nextPageElements.get(0); return baseUrl + hrefElement.attributes().get("href"); } return "";}
Тогда основной метод будет таким. Формируется начальная строка поиска. Закачиваются данные по одной странице. Если есть следующая страница, то идет переход к ней. По окончании передаются накопленные данные.
@Overridepublic Collection<Movie> scrap() { String url = String.format( baseUrl + "/search/title/?release_date=%s,%s&countries=%s", startDate, endDate, countries ); List<Movie> items = new ArrayList<>(); String nextUrl = url; while (true) { nextUrl = scrap(nextUrl, items); if ("".equals(nextUrl)) { break; } try { Thread.sleep(50); } catch (InterruptedException e) { } } return items;}
Подробности по остальным методам можно найти в ссылках на ресурсы.
Отправка данных в топик
Формируется следующий сервис: MovieProducer. Здесь будет один единственный публичный метод: run.
Создается продюсер для кафки. Загружаются данные из источника. Трансформируются и отправляются в топик.
public void run() { try (SimpleStringStringProducer producer = new SimpleStringStringProducer( bootstrapServers, clientId, topic)) { Collection<Data.Movie> movies = movieDirectScrapingService.scrap(); List<SimpleStringStringProducer.KeyValueStringString> kvList = new ArrayList<>(); for (Data.Movie move : movies) { Map<String, String> map = new HashMap<>(); map.put("title_id", move.titleId); map.put("title_url", move.titleUrl); String value = JSONObject.toJSONString(map); String key = UUID.randomUUID().toString(); kvList.add(new SimpleStringStringProducer.KeyValueStringString(key, value)); } producer.produce(kvList); }}
Теперь все вместе
Формируются нужные параметры для поиска. Загружаются данные и
отправляются в топик.
Для этого понадобится еще один класс: MovieDirectScrapingExecutor.
С одним публичным методом: run.
В цикле создаются данные для поиска из текущей даты. Происходит загрузка и отправка данных в топик.
public void run() { int countriesCounter = 0; List<String> countriesSource = Arrays.asList("us"); while (true) { try { LocalDate localDate = LocalDate.now(); int year = localDate.getYear(); int month = localDate.getMonthValue(); int day = localDate.getDayOfMonth(); String monthString = month < 9 ? "0" + month : Integer.toString(month); String dayString = day < 9 ? "0" + day : Integer.toString(day); String startDate = year + "-" + monthString + "-" + dayString; String endDate = startDate; String language = "en"; String countries = countriesSource.get(countriesCounter); execute(language, startDate, endDate, countries); Thread.sleep(1000); countriesCounter += 1; if (countriesCounter >= countriesSource.size()) { countriesCounter = 0; } } catch (InterruptedException e) { } }}
Для запуска потребуется экземпляр класса MovieDirectScrapingExecutor, который можно запустить с нужными параметрами, к примеру, из метода main.
Пример отправляемых данных для одного фильма.
{ "base_name_url": "https:\/\/www.imdb.com\/name", "participant_ids": "nm7947173~nm2373827~nm0005288~nm0942193~", "title_id": "tt13121702", "rating": "0.0", "base_url": "https:\/\/www.imdb.com", "description": "It's Christmas time and Jackie (Carly Hughes), an up-and-coming journalist, finds that her life is at a crossroads until she finds an unexpected opportunity - to run a small-town newspaper ... See full summary ", "runtime": "", "title": "The Christmas Edition", "director_ids": "nm0838289~", "title_url": "\/title\/tt13121702\/?ref_=adv_li_tt", "director_names": "Peter Sullivan~", "genres": "Drama, Romance", "base_title_url": "https:\/\/www.imdb.com\/title", "participant_names": "Carly Hughes~Rob Mayes~Marie Osmond~Aloma Wright~"}
Подробности можно найти в ссылках на ресурсы.
Тесты
Для тестирования основной логики, которая связана с отправкой
данных, можно воспользоваться юнит-тестами. В тестах предварительно
создается kafka-сервер.
См. Apache Kafka и тестирование с Kafka Server.
Сам тест: MovieProducerTest.
public class MovieProducerTest { @Test void simple() throws InterruptedException { String brokerHost = "127.0.0.1"; int brokerPort = 29092; String zooKeeperHost = "127.0.0.1"; int zooKeeperPort = 22183; String bootstrapServers = brokerHost + ":" + brokerPort; String topic = "q-data"; String clientId = "simple"; try (KafkaServerService kafkaServerService = new KafkaServerService( brokerHost, brokerPort, zooKeeperHost, zooKeeperPort ) ) { kafkaServerService.start(); kafkaServerService.createTopic(topic); MovieDirectScrapingService movieDirectScrapingServiceImpl = () -> Collections.singleton( new Data.Movie() ); MovieProducer movieProducer = new MovieProducer(bootstrapServers, clientId, topic, movieDirectScrapingServiceImpl); movieProducer.run(); kafkaServerService.poll(topic, "simple", 1, 5, (records) -> { assertTrue(records.count() > 0); ConsumerRecord<String, String> record = records.iterator().next(); JSONParser jsonParser = new JSONParser(); JSONObject jsonObject = null; try { jsonObject = (JSONObject) jsonParser.parse(record.value()); } catch (ParseException e) { e.printStackTrace(); } assertNotNull(jsonObject); }); Thread.sleep(5000); } }}
Заключение
Конечно, описанный здесь способ получения источника потоковых данных, строго потоковым не является. Но для исследований и прототипов вполне может сойти.