Как обеспечить быструю и надежную миграцию наборов данных на серверную часть fiware? - PullRequest
Новая
       21

Как обеспечить быструю и надежную миграцию наборов данных на серверную часть fiware?

1 голос
/ 28 мая 2019

Передача очень больших наборов данных в fiware бэкэнд: SourceDB ->Orion -> Cygnus -> Postgres.Для этого я пишу скрипт Python, извлекающий строки, и для каждой выбранной строки отправляю полезную нагрузку в Orion.

Скрипт запускается с интервалами 150 мс (0,15 с), но, к моему удивлению, после примерно 10 итераций, толькодва значения (первый и последний отправленные полезные данные) были сохранены в приемнике Postgres.Это означает, что 80% наборов данных не сохраняются в приемнике.

Сценарий: 

import psycopg2
from time import sleep
from config import config
from tqdm import tqdm
import requests
import json

def val_json():
    db = "select to_json(d) from (  select \
        n.noise_data as measurand, \
        n.factor as \"sonometerClass\", \
        to_timestamp(n.seconds) as \"dateObserved\", \
        l.description as name, \
            json_build_object( \
                'coordinates', \
                json_build_array(l.node_lon, l.node_lat) \
            ) as location \
        from noise as n \
            inner join deployment as d on \
                d.deployment_id = n.deployment_id \
            inner join location as l on \
                l.location_id = d.location_id \
        order by n.seconds asc \
    ) as d"
    return db

def main():

    url = 'http://localhost:1026/v2/entities/002/attrs?options=keyValues'
    headers = {"Content-Type": "application/json", \
               "fiware-service": "urbansense",  \
               "fiware-servicepath": "/basic"}

    conn = None
    try:
        params = config()
        with psycopg2.connect(**params) as conn:
            with conn.cursor(name='my_cursor') as cur:
                cur.itersize = 5000
                cur.execute(val_json())

                for row in tqdm(cur):
                    jsonData = json.dumps(row)
                    if jsonData.startswith('[') and jsonData.endswith(']'):
                        jsonData = jsonData[1:-1]
                        print()
                        print(jsonData)
                    requests.post(url, data= jsonData, headers=headers)
                    sleep(0.15)

                cur.close()
    except (Exception, psycopg2.DatabaseError) as error:
        print(error)
    finally:
        if conn is not None:
            conn.close()

if __name__ == '__main__':
    main()

Первые десять итераций (полезные нагрузки): 

$python3 noiselevelObserved.py
0it [00:00, ?it/s]
{"measurand": 64.8, "sonometerClass": 1, "dateObserved": "1970-01-01T01:00:15+01:00", "name": "Trindade", "location": {"coordinates": [-8.609973, 41.151943]}}
1it [00:00,  1.75it/s]
{"measurand": 58.8, "sonometerClass": 0, "dateObserved": "1970-01-01T01:00:16+01:00", "name": "Trindade", "location": {"coordinates": [-8.609973, 41.151943]}}
2it [00:00,  2.23it/s]
{"measurand": 56.5, "sonometerClass": 0, "dateObserved": "1970-01-01T01:00:17+01:00", "name": "Trindade", "location": {"coordinates": [-8.609973, 41.151943]}}
3it [00:00,  2.76it/s]
{"measurand": 61.1, "sonometerClass": 1, "dateObserved": "1970-01-01T01:00:18+01:00", "name": "Casa da Musica", "location": {"coordinates": [-8.63041, 41.158091]}}
4it [00:01,  3.31it/s]
{"measurand": 108.5, "sonometerClass": 2, "dateObserved": "1970-01-01T01:00:18+01:00", "name": "Pr. Liberdade Cardosas", "location": {"coordinates": [-8.611119, 41.146023]}}
5it [00:01,  3.86it/s]
{"measurand": 56.5, "sonometerClass": 0, "dateObserved": "1970-01-01T01:00:18+01:00", "name": "Trindade", "location": {"coordinates": [-8.609973, 41.151943]}}
6it [00:01,  4.35it/s]
{"measurand": 59.9, "sonometerClass": 1, "dateObserved": "1970-01-01T01:00:19+01:00", "name": "Casa da Musica", "location": {"coordinates": [-8.63041, 41.158091]}}
7it [00:01,  4.78it/s]
{"measurand": 97.2, "sonometerClass": 2, "dateObserved": "1970-01-01T01:00:19+01:00", "name": "D. Manuel II", "location": {"coordinates": [-8.625192, 41.148558]}}
8it [00:01,  5.12it/s]
{"measurand": 108.6, "sonometerClass": 2, "dateObserved": "1970-01-01T01:00:19+01:00", "name": "Pr. Liberdade Cardosas", "location": {"coordinates": [-8.611119, 41.146023]}}
9it [00:01,  5.41it/s]
{"measurand": 57.1, "sonometerClass": 0, "dateObserved": "1970-01-01T01:00:19+01:00", "name": "Trindade", "location": {"coordinates": [-8.609973, 41.151943]}}
10it [00:02,  5.63it/s]
{"measurand": 53.9, "sonometerClass": 0, "dateObserved": "1970-01-01T01:00:20+01:00", "name": "Casa da Musica", "location": {"coordinates": [-8.63041, 41.158091]}}

Значения чтения атрибутов сохраняются в postgres: 

postgres=# select  * from urbansense.basic_002_noiselevelobserved ;
  recvtimets   |         recvtime         | fiwareservicepath | entityid |     entitytype     |    attrname    |    attrtype     |               attrvalue               | attrmd
---------------+--------------------------+-------------------+----------+--------------------+----------------+-----------------+---------------------------------------+--------
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | dateObserved   | Text            | 1970-01-01T01:00:15+01:00             | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | latitude       | Number          | 41.1591                               | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | location       | StructuredValue | {"coordinates":[-8.609973,41.151943]} | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | longitude      | Number          | -8.65915                              | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | measurand      | Number          | 64.8                                  | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | name           | Text            | Trindade                              | []
 1559045918129 | 2019-05-28T12:18:38.129Z | /basic            | 002      | NoiseLevelObserved | sonometerClass | Number          | 1                                     | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | dateObserved   | Text            | 1970-01-01T01:00:20+01:00             | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | latitude       | Number          | 41.1591                               | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | location       | StructuredValue | {"coordinates":[-8.63041,41.158091]}  | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | longitude      | Number          | -8.65915                              | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | measurand      | Number          | 53.9                                  | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | name           | Text            | Casa da Musica                        | []
 1559045919723 | 2019-05-28T12:18:39.723Z | /basic            | 002      | NoiseLevelObserved | sonometerClass | Number          | 0                                     | []
(14 rows)

Изменение скорости стрельбы на интервал в 1 секунду не дает большого улучшения, только 3 набора результатов (полезные нагрузки) были сохранены (потеря 70%): 

postgres=# select  * from urbansense.basic_002_noiselevelobserved ;
  recvtimets   |         recvtime         | fiwareservicepath | entityid |     entitytype     |    attrname    |    attrtype     |               attrvalue               | attrmd
---------------+--------------------------+-------------------+----------+--------------------+----------------+-----------------+---------------------------------------+--------
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | dateObserved   | Text            | 1970-01-01T01:00:15+01:00             | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | latitude       | Number          | 41.1591                               | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | location       | StructuredValue | {"coordinates":[-8.609973,41.151943]} | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | longitude      | Number          | -8.65915                              | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | measurand      | Number          | 64.8                                  | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | name           | Text            | Trindade                              | []
 1559046840569 | 2019-05-28T12:34:00.569Z | /basic            | 002      | NoiseLevelObserved | sonometerClass | Number          | 1                                     | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | dateObserved   | Text            | 1970-01-01T01:00:18+01:00             | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | latitude       | Number          | 41.1591                               | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | location       | StructuredValue | {"coordinates":[-8.609973,41.151943]} | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | longitude      | Number          | -8.65915                              | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | measurand      | Number          | 56.5                                  | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | name           | Text            | Trindade                              | []
 1559046845620 | 2019-05-28T12:34:05.620Z | /basic            | 002      | NoiseLevelObserved | sonometerClass | Number          | 0                                     | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | dateObserved   | Text            | 1970-01-01T01:00:20+01:00             | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | latitude       | Number          | 41.1591                               | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | location       | StructuredValue | {"coordinates":[-8.63041,41.158091]}  | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | longitude      | Number          | -8.65915                              | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | measurand      | Number          | 53.9                                  | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | name           | Text            | Casa da Musica                        | []
 1559046850679 | 2019-05-28T12:34:10.679Z | /basic            | 002      | NoiseLevelObserved | sonometerClass | Number          | 0                                     | []
(21 rows)

Честно говоря, мне не понравилась бы скорость стрельбыбыть до в секунду из-за размера данных, которые необходимо перенести для продолжения моего исследования.Когда я впервые попробовал его с интервалом в 1 секунду, я понял, что на завершение уйдут месяцы (возможно, 4).

Вопрос: Это Orion CB (или, возможно, Cygnus)не создан, чтобы обладать такой способностью принимать значения сущностей / атрибутов с такой скоростью (150 мс / 0,15 с), или, может быть, Cygnus не достаточно умен для получения уведомления от Orion с такой скоростью?.

Буду признателен за любые предложения по сохранению всех значений в кратчайшие сроки.

1 Ответ

2 голосов
/ 30 мая 2019

Исходя из обсуждения в комментариях к вопросу, решение этой проблемы заключалось в удалении параметра throttling в подписке.Это имеет смысл: регулирование приводит к тому, что некоторые уведомления не отправляются (в данном конкретном случае 80% всех уведомлений).

...