Часть 7 : Настройка сервера Node.js для получения данных API и отправки данных в Elasticsearch

Оглавление | Читать далее: Часть 8: Создание клиентской части приложения с помощью React

В части 6 мы создали конвейер ввода данных (earthquake_data_pipeline) для преобразования данных, полученных из API USGS. Мы также создали индекс earthquakes с нужным отображением.

В этом блоге мы собираемся настроить сервер для:

• получать данные о землетрясениях из USGS API
• отправлять полученные данные в Elasticsearch earthquake_data_pipeline для преобразования данных
• дать указание Elasticsearch внести преобразованные данные в индекс earthquakes.

Следующая диаграмма обобщает задачи, которые мы будем выполнять в этом блоге.

Ресурсы

Вы предпочитаете посмотреть видео для изучения этого материала? Нажмите на ссылку ниже!

• Эпизод 7: Настройка сервера Node.js для получения данных API и отправки данных в Elasticsearch

Хотите получить код, рассмотренный в этом блоге? Нажмите на ссылку ниже, чтобы получить к нему доступ!

• GitHub репозиторий для части 7

Распознавание

Я хочу выразить огромную благодарность Марку Мэйфилду!

Когда я впервые пытался понять, как получать и вводить данные API в Elasticsearch, я наткнулся на его блог.

Его блог помог мне написать часть 7. Спасибо, Марк!

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

Я также разбил код построчно для более легкого понимания.

Получение данных API

Шаг 1: Настройте свой сервер для получения данных о землетрясениях из USGS API

Откройте earthquake_app с помощью редактора кода по вашему выбору.

В каталоге server создайте каталог data_management. В директории data_management создайте файл retrieve_and_ingest_data.js (красная рамка).

Файл retrieve_and_ingest_data.js будет предписывать серверу выполнять три действия:

1. после получения http-запроса от браузера получить данные из USGS API
2. отправить полученные данные в Elasticsearch earthquake_data_pipeline для преобразования данных
3. проинструктировать Elasticsearch о включении преобразованных данных в индекс earthquakes.

В файле retrieve_and_ingest_data.js скопируйте и вставьте следующий фрагмент.

//in server/data_management/retrieve_and_ingest_data.js
const express = require('express');
const router = express.Router();
const axios = require('axios');
const client = require('../elasticsearch/client');
require('log-timestamp');

const URL = `https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.geojson`;

router.get('/earthquakes', async function (req, res) {
  console.log('Loading Application...');
  res.json('Running Application...');

  indexData = async () => {
    try {
      console.log('Retrieving data from the USGS API');

      const EARTHQUAKES = await axios.get(`${URL}`, {
        headers: {
          'Content-Type': ['application/json', 'charset=utf-8'],
        },
      });

      console.log('Data retrieved!');

      results = EARTHQUAKES.data.features;

      console.log('Indexing data...');

      results.map(
        async (results) => (
          (earthquakeObject = {
            place: results.properties.place,
            time: results.properties.time,
            tz: results.properties.tz,
            url: results.properties.url,
            detail: results.properties.detail,
            felt: results.properties.felt,
            cdi: results.properties.cdi,
            alert: results.properties.alert,
            status: results.properties.status,
            tsunami: results.properties.tsunami,
            sig: results.properties.sig,
            net: results.properties.net,
            code: results.properties.code,
            sources: results.properties.sources,
            nst: results.properties.nst,
            dmin: results.properties.dmin,
            rms: results.properties.rms,
            mag: results.properties.mag,
            magType: results.properties.magType,
            type: results.properties.type,
            longitude: results.geometry.coordinates[0],
            latitude: results.geometry.coordinates[1],
            depth: results.geometry.coordinates[2],
          }),
          await client.index({
            index: 'earthquakes',
            id: results.id,
            body: earthquakeObject,
            pipeline: 'earthquake_data_pipeline',
          })
        )
      );

      if (EARTHQUAKES.data.length) {
        indexData();
      } else {
        console.log('Data has been indexed successfully!');
      }
    } catch (err) {
      console.log(err);
    }

    console.log('Preparing for the next round of indexing...');
  };
  indexData();
});

module.exports = router;

Давайте рассмотрим этот фрагмент кода.

Внимание!

Для справки я привел скриншоты кода, который я буду объяснять.

Если вам нужно скопировать и вставить код, пожалуйста, обратитесь к фрагменту кода выше или к репозиторию GitHub для части 7.

Строки 1-3 требуют зависимостей, необходимых для получения и отправки HTTP-запросов.

Строка 4
Требуется Elasticsearch Client, который мы создали в части 4.

Строка 5
Нам требуется зависимость log-timestamp, которую мы установили в части 2. Она добавляет временные метки к сообщениям, отображаемым в терминале с помощью метода console.log().

Вы увидите эту зависимость, когда мы будем получать данные из USGS API.

Строка 7
Мы создаем константу URL и устанавливаем ее равной URL всех данных о землетрясениях за последние 30 дней API. Мы будем извлекать данные из этого API.

Мы используем строки 9-77 для определения того, что должен делать наш сервер, когда получает HTTP-запрос к конечной точке /earthquakes.

Строка 9
Мы определяем маршрут для /earthquakes.

Строка 10
Когда сервер получает HTTP-запрос к конечной точке /earthquakes, он выводит в терминале сообщение «Loading application…».

Строка 11
В браузере (клиент) выводится сообщение «Running Application…».

Мы используем строки 13-77 для:

• получить данные из API USGS
• создать объект для каждого землетрясения
• отправить объекты в Elasticsearch earthquake_data_pipeline для преобразования данных
• проинструктировать Elasticsearch о вводе преобразованных объектов в индекс earthquakes.

Строка 13

Мы определяем функцию indexData().

Строка 15
Когда эта функция вызывается, она печатает в терминале сообщение «Retrieving data from the USGS API».

Строки 17-21
Мы отправляем GET-запрос в USGS API.

Строка 23
Мы выводим сообщение «Данные получены!» в терминале.

Строка 25
Мы создаем переменную results. Эта переменная равна полученным от API данным о землетрясениях.

Помните структуру данных USGS API, рассмотренную в части 5?

EARTHQUAKES.data.features дает вам доступ к массиву features (зеленый квадрат).

Этот массив содержит объекты землетрясений. Каждый объект содержит информацию об одном землетрясении.

Внутри каждого объекта поля properties (оранжевая рамка) и geometry (синяя рамка) содержат нужную нам информацию (розовая рамка).

Строка 27
Выводим в терминале сообщение «Indexing data».

Строки 29-55

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

Строки 56-61

Мы используем наш экземпляр Elasticsearch Client для:

• отправить полученные данные API в Elasticsearch earthquake_data_pipeline для преобразования данных
• инструктировать Elasticsearch о вводе преобразованных данных в индекс earthquakes.

Строка 56
Мы используем метод client.index() для индексации преобразованных данных.

Строка 57
Мы указываем, что данные должны быть проиндексированы в индекс earthquakes.

Строка 58
Мы присваиваем каждому документу землетрясения идентификатор, идентичный идентификатору объекта землетрясения, полученного из API. Мы делаем это для того, чтобы id землетрясения из API USGS и id документа совпадали.

Строка 59
Мы устанавливаем body равным earthquakeObject, который мы определили в строках 31-55. body представляет документ одного землетрясения.

Строка 60
Мы инструктируем Elasticsearch client отправить полученные данные в Elasticsearch earthquake_data_pipeline.

Конвейер earthquake_data_pipeline преобразует данные так, как мы указали в части 6.

Затем Elasticsearch проиндексирует преобразованные данные в индекс earthquakes.

Строки 65-72
Мы создаем оператор if, else, catch.

Строки 65-67
Если остались данные, которые не были проиндексированы, продолжайте вызывать функцию indexData().

Строки 67-69
Если все данные были проиндексированы, то выведите в терминал сообщение «Data has been indexed successfully!».

Строки 70-72
Если во время индексирования произошла ошибка, выведите ее в терминал.

Строка 74
После завершения индексирования данных выведите в терминале сообщение «Подготовка к следующему раунду индексирования…».

Строка 76
Вызываем функцию indexData().

Строка 79
Мы раскрываем маршрутизатор через модуль Node.js exports, так как он будет использоваться в server.js.

Добавляем маршрут приема данных в server.js

Шаг 1: откройте server.js в каталоге server.

Скопируйте и вставьте следующее в строки 8 и 10.

//in server/server.js
const data = require('./data_management/retrieve_and_ingest_data');

app.use('/ingest_data', data);

Ваш server.js должен выглядеть следующим образом:

Давайте пройдемся по этому фрагменту.

Строка 8
Мы создаем константу data и требуем файл retrieve_and_ingest_data.js из нашей директории data_management.

Строка 10
Когда сервер получает http-запрос от url-пути, начинающегося с /ingest_data, выполняется функция, которую мы определили в retrieve_and_ingest_data.js.

Давайте проверим, работает ли это!

Используя терминал, перейдите в каталог earthquake_app. Запустите сервер, выполнив следующую команду.

//in terminal within the earthquake_app directory
npm start

В терминале вы должны увидеть что-то похожее на скриншот.

Откройте новый браузер. Расположите терминал и браузер рядом друг с другом, как показано ниже.

Скопируйте и вставьте следующий url в адресную строку браузера и нажмите enter:

//in the address bar of your browser
http://localhost:3001/ingest_data/earthquakes

В терминале вы увидите следующее сообщение, подтверждающее, что данные получены, преобразованы и введены в Elasticsearch. Вы также увидите, что к сообщениям, выводимым на консоль, была добавлена временная метка!

Вы также увидите сообщение «Running Application…», отображаемое в браузере.

Оценка результатов преобразования и ввода данных

Давайте посмотрим, правильно ли были преобразованы и введены данные в Elasticsearch.

Одним из способов просмотра данных, хранящихся в Elasticsearch, является инструмент Kibana Discover.

С помощью Discover вы можете быстро искать и фильтровать данные, получать информацию о структуре полей и отображать результаты в виде визуализации. Вы также можете настраивать и сохранять свои поиски и размещать их на приборной панели (выдержка из документации Kibana).

Чтобы использовать Discover, мы должны сначала указать, где он может найти данные, которые мы хотим исследовать.

Для этого мы создаем представление данных.

Представление данных data view позволяет вам указать источник данных, который вы хотите исследовать, чтобы Kibana могла найти эти данные.

На главной странице Kibana нажмите на опцию Stack Management (красная рамка).

На странице Stack Management нажмите на опцию Data Views (красная рамка).

Нажмите на кнопку Создать представление данных (красная рамка).

Вы увидите, что индекс earthquakes (красная рамка) указан как один из источников данных.

В разделе Name введите название индекса earthquakes(красная рамка).

Discover также позволяет исследовать данные в различных временных диапазонах.

Поскольку наши данные о землетрясениях находятся во временном ряду, мы сможем использовать функцию временного диапазона в Discover.

Поскольку данные о землетрясениях имеют поле временной метки @timestamp, Kibana автоматически обновит поле Timestamp полем @timestamp (синее поле).

Нажмите на опцию Создать представление данных (зеленая рамка).

Вы увидите, что представление данных под названием earthquakes было создано. Поле Time было установлено в поле @timestamp.

Нажмите на значок меню (красный квадрат) и выберите опцию Discover (синий квадрат).

Вы увидите, что представление данных earthquakes (красная рамка) уже выбрано на странице Discover.

На экране также отображается сообщение «Нет результатов, соответствующих вашим критериям поиска» (оранжевая рамка).

Давайте посмотрим, почему мы столкнулись с этим сообщением.

Взгляните на временной диапазон, выделенный зеленым квадратом.

Discover в настоящее время получает указание отображать данные, которые были собраны в течение последних 15 минут. Поскольку наш индекс не содержит таких данных, мы получаем сообщение о том, что ни один результат не соответствует нашим критериям поиска.

Давайте настроим нашу временную шкалу на Последние 30 дней, чтобы просмотреть результаты.

Нажмите на значок календаря (синее поле). Вы увидите выпадающее меню, в котором можно выбрать временной диапазон, в котором вы хотите просмотреть данные. Выберите опцию Последние 30 дней (красная рамка).

Вы увидите, что в индексе earthquakes имеется 9 108 документов (красная рамка), которые были проиндексированы из USGS API.

В таблице (желтая рамка) отображаются документы в индексе earthquakes. В каждой строке отображается информация о документе (землетрясении).

Давайте посмотрим на документ, чтобы проверить, правильно ли были преобразованы данные перед тем, как попасть в Elasticsearch.

Нажмите на стрелку (красное поле), чтобы развернуть информацию о документе (синее поле).

Вы увидите, что документ содержит нужные поля и формат временной метки, который мы установили в части 6.

В этом блоге мы успешно

• извлекли данные о землетрясениях из USGS API и отправили их в Elasticsearch
• преобразовали полученные данные с помощью конвейера ввода данных (earthquake_data_pipeline)
• включили преобразованные данные в индекс earthquakes.

Теперь, когда у нас есть данные в Elasticsearch, пришло время настроить наше приложение для поиска.

Переходим к части 8 для создания клиента на React, чтобы пользователи могли указать данные о землетрясениях, которые они ищут!