Введение
Для анализа потоковых данных необходимы источники этих данных. Так же важна сама информация, которая предоставляется источниками. А источники с текстовой информацией, к примеру, еще и редки.
Из интересных источников можно выделить следующие: 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 "";
}Тогда основной метод будет таким. Формируется начальная строка поиска. Закачиваются данные по одной странице. Если есть следующая страница, то идет переход к ней. По окончании передаются накопленные данные.
@Override
public 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);
}
}
}Заключение
Конечно, описанный здесь способ получения источника потоковых данных, строго потоковым не является. Но для исследований и прототипов вполне может сойти.
