Cómo analizar datos de series temporales con Python e InfluxDB

Python e InfluxDB

datos de series de tiempo Hoy proviene de muchas fuentes. Es posible que las bases de datos relacionales tradicionales no manejen bien los datos de series temporales porque:

• Cada fuente de datos requiere un esquema personalizado. Este diseño significa que tiene que dedicar más tiempo a decidir cómo almacenar sus datos. Además, si cambia la fuente de datos subyacente, es posible que deba cambiar el esquema de la tabla.
• Las bases de datos relacionales tradicionales no caducan los datos. Por lo general, desea mantener los datos de series temporales solo durante un período de tiempo y descartar automáticamente el resto.
• La nueva base de datos de series temporales optimiza el formato de almacenamiento para que varios clientes puedan escribir datos más rápido y también leer los datos más recientes más rápido. Las bases de datos relacionales están más preocupadas por la duplicación de datos y la contención de bloqueo de filas.

[ Top considerations for cloud-native databases and data analytics. ]

En este tutorial te mostraré cómo usar Entrada en la base de datos, una plataforma de series temporales de código abierto.Me gusta porque proporciona integración con otras herramientas listas para usar (incluyendo Grafana y pitón 3), que utiliza Flux, un lenguaje potente y sencillo para ejecutar consultas.

requisitos previos

Este tutorial requiere:

• Una instalación de Docker o Podman para ejecutar InfluxDB.También puede hacer una instalación completa, pero no la cubriré aquí.
• InfluxDB 2.4.0 o posterior
• flujo para consulta (en lugar de Flux QL)
• Una distribución de Linux (yo uso Fedora Linux)
• Python 3 y algo de experiencia en secuencias de comandos
• El conocimiento básico de bases de datos relacionales como MariaDB puede ser útil, pero no se requiere Flux

Ejecutar el servidor InfluxDB desde un contenedor

Los contenedores son probablemente la forma más fácil de comenzar.Use volúmenes externos para conservar los datos en los reinicios y actualizaciones del contenedor (consulte la página del contenedor para ver todas las opciones posibles):

$ podman pull influxdb:latest
$ podman run --detach --volume /data/influxdb:/var/lib/influxdb --volumne /data:/data:rw  --name $ $ $ influxdb_raspberrypi --restart always --publish 8086:8086 influxdb:latest
$ podman logs --follow influxdb_raspberrypi

asignar un nombre /data Importe algunos archivos CSV más tarde dentro del contenedor.

A continuación, vaya a la máquina que ejecuta InfluxDB (como http://localhost:8086) y Complete los pasos de instalación:

1. Crear un nuevo usuario.
2. Crear balde (equivalente a una tabla de base de datos relacional) no caducada, ¿dónde colocaría sus datos? Se llama USTS (abreviatura de Tanque de Almacenamiento Subterráneo) por las razones que verá a continuación.

3. Cree un token de API de lectura/escritura solo para el depósito USTS.

El token de la API será similar al siguiente:

nFIaywYgg0G6oYtZCDZdp0StZqYmz4eYg23KGKcr5tau05tb7T2DY-LQSIgRK66QG9mkkKuK2nNCoIKJ0zoyng==

Mantenga su token seguro, ya que lo usará para leer y escribir datos en su depósito USTS.

Trabajar con conjuntos de datos

En este tutorial, usaré el Portal de datos abiertos de Connecticut.

Específicamente, usaré Tanques de almacenamiento subterráneo (UST): detalles de las instalaciones y del tanque de almacenamiento datos(descargar). Las Regulaciones de Tanques de Almacenamiento Subterráneos y el Programa de Cumplimiento de Tanques de Almacenamiento Subterráneos de Connecticut han estado vigentes desde noviembre de 1985. Esta lista se basa en la información de notificación enviada por el público desde 1985 y se actualiza semanalmente.

Las razones por las que me gusta este conjunto de datos son las siguientes:

• Este es un conjunto de datos relativamente extenso con 49.1K filas y 27 columnas. Esto requerirá algunas técnicas de normalización de datos y de importación de big data.
• Tiene datos en forma de serie temporal (columna de fecha de última utilización).
• También tiene detalles geográficos (coordenadas de latitud y longitud) que se pueden usar para ejecutar algunas consultas interesantes usando la función de experimento de geolocalización.

puedes usarlo curl. P.ej:

$ curl --location --silent --fail --show-error --output ~/Downloads/ust.csv '

$ wc -l ~/Downloads/ust.csv
49090 /home/josevnz/Downloads/ust.csv

A continuación, le mostraré cómo obtener sus datos en su depósito y algunos de los problemas con los que podría encontrarse.

[ Download a sysadmin’s guide to Bash scripting. ]

Datos de importacion

Antes de decidir qué importar y qué ignorar, es una buena idea definir preguntas que puedan responderse con datos. Aquí hay algunos ejemplos de lo que este tutorial intentará responder:

• Número total de tanques informados (activos, inactivos)
• Número de tanques inactivos por tiempo de instalación
• Número de diferentes tipos de tanques, agrupados por tipo de sustancia
• Número de tanques cerca de Hartford, Connecticut

Verifique las columnas disponibles y observe qué ignorar durante el proceso de importación para responder las preguntas que defina.

Ver sus datos:

ST Site ID Number,Site Name,Site Address,Site City,Site Zip,Tank No.,Status of Tank,Compartment,Estimated Total Capacity (gallons),Substance Currently Stored,Last Used Date,Closure Type,Construction Type - Tank,Tank Details,Construction Type - Piping,Piping Details,Installation Date,Spill Protection,Overfill Protection,Tank Latitude,Tank Longitude,Tank Collection Method,Tank Reference Point Type,UST Site Latitude,UST Site Longitude,Site Collection Method,Site Reference Point Type
50-11456,Brewer Dauntless Marina,9 NOVELTY LN,ESSEX,06426,1,Permanently Closed,,4000,Gasoline,10/18/2018,Tank was Removed From Ground,Coated & Cathodically Protected Steel (sti-P3),Double Walled,Flexible Plastic,"Containment Sumps @ Dispensers,Containment Sumps @ Tanks,Double Walled,Metallic fittings isolated from soil and water",06/01/1999,Spill Bucket,Ball Float Device,41.350018,-72.385442,Address Matching,Approximate Location,41.350018,-72.385442,Address Matching,Approximate Location
106-1923,FOOD BAG #509,1652 BOSTON POST RD,OLD SAYBROOK,06475,D1,Permanently Closed,a,10000,Diesel,03/01/1998,Tank was Removed From Ground,Coated & Cathodically Protected Steel (sti-P3),,Rigid Fiberglass Reinforced Plastic,,02/01/1983,,,41.286115,-72.414762,Address Matching,Approximate Location,41.286115,-72.414762,Address Matching,Approximate Location

Puede usar el importador masivo InfluxDB como primer uso usando el protocolo de línea. Esto significa que puede ayudar a InfluxDB a digerir datos como este:

1. Dígale qué acción está tomando.
2. Seleccione el tipo apropiado para Tipo de datos, como Número, Fecha o Coordenadas geográficas.
3. Seleccione las columnas para omitir.

usar Comentarios CSV es una forma, pero limita cuánto puede hacer durante el proceso de importación.

Luego, debe decidir dónde y cómo almacenar los datos. Conceptos utilizados por InfluxDB Etiquetas, campos y medidas:

• Los cubos son bases de datos que almacenan datos.
• InfluxDB almacena datos en métricas (equivalentes a tablas en bases de datos relacionales).esto es Barriles de gasolina.
• Una etiqueta es una combinación de clave y valor utilizada en el índice. Sus valores no cambian con el tiempo. Piense en ellos como metadatos. en este caso, Sustancias actualmente almacenadas y tipo de cierre es la etiqueta.
• Los campos cambian con el tiempo y no están indexados.Este Capacidad total estimada (galones) total es un contador que cambia con el tiempo.
• El tiempo es la estructura de los datos y provendrá del período del informe, no del campo de fecha actualizado.

Tenga en cuenta que no una sino dos columnas de fecha y hora:

Y, como toda serie temporal, Solo puede haber una columna de fecha y hora.

¿Entonces lo que hay que hacer? Bueno, puede dividir los datos en dos cubos según lo que desee rastrear, almacenar uno de ellos como una etiqueta (inútil porque es más difícil trabajar con él) o ignorarlo por completo.

Para este análisis, le importa más la última fecha de uso, así que ignore la fecha de instalación.

¿Cómo encontrar datos?

#constant measurement,fuel_tanks
#datatype ignore,ignore,ignore,tag,ignore,ignore,tag,tag,long,tag,dateTime:01/02/2006,tag,tag,ignore,tag,ignore,ignore,tag,tag,double,double,ignored,ignored,ignored,ignored,ignored,ignored
ID,Name,Address,City,Zip,TankNo,Status,Compartment,EstimatedTotalCapacity,SubstanceStored,LastUsed,ClosureType,ConstructionType,Details,ConstructionType,PipingDetails,InstallationDate,SpillProtection,OverfillProtection,Latitude,Longitude,CollectionMethod,ReferencePointType,USTLatitude,USTLongitude,CollectionMethod,ReferencePointType

El comando completo se ve así (importar ust.sh):

$ /usr/bin/podman run --interactive --tty --volume "$header_file:/data/headers.csv" --volume "$csv_file:/data/tanks.csv" influxdb influx write "$dryrun" --bucket $BUCKET --org $ORG --format csv --skipHeader=1 --url "$url" --file "/data/headers.csv" --file "/data/tanks.csv"

Si lo ejecuta en modo seco, podrá ver el protocolo de línea utilizado para importar los datos:

fuel_tanks,City=BRISTOL,ConstructionType=Flexible\ Plastic,OverfillProtection=Audible\ Alarm,SpillProtection=Spill\ Bucket,Status=Currently\ In\ Use,SubstanceStored=Gasoline EstimatedTotalCapacity=15000i,Latitude=41.65641,Longitude=-72.91408
fuel_tanks,City=Hartford,ConstructionType=Flexible\ Plastic,OverfillProtection=Audible\ Alarm,SpillProtection=Spill\ Bucket,Status=Currently\ In\ Use,SubstanceStored=Diesel EstimatedTotalCapacity=1000i,Latitude=41.75538,Longitude=-72.680618
fuel_tanks,City=BERLIN,ConstructionType=Flexible\ Plastic,OverfillProtection=Audible\ Alarm,SpillProtection=Spill\ Bucket,Status=Currently\ In\ Use,SubstanceStored=Gasoline EstimatedTotalCapacity=10000i,Latitude=41.646417,Longitude=-72.729937

...

Pero incluso con esta flexibilidad, a veces hay que escribir importadores personalizados.

Cuando los comentarios CSV son insuficientes

Entrada en la base de datos write Los comandos son muy flexibles, pero los datos presentan algunos desafíos que la herramienta no puede resolver:

1. Los datos de esta colección no están ordenados por fecha de instalación. Esto afecta el rendimiento de la consulta, por lo que debe ordenar los datos antes de insertarlos.afluencia write El comando asume (correctamente) que los datos aparecen ordenados.
2. La ciudad a veces está en mayúsculas, pero no de manera consistente.
3. Falta la última fecha utilizada en algunos puntos de datos. Para este ejercicio, suponga que la fecha del último uso es igual a la fecha actual si el tanque todavía está en uso.
4. Algunos tanques fuera de servicio solo tienen códigos postales y no tienen coordenadas.No hay problema, usarás código postal Obtener latitud y longitud.

Escribí un script importador personalizado llamado import_ust.py Manejar estos casos con gracia.

[ Want to test your sysadmin skills? Take a skills assessment today. ]

ejecutar algunas consultas

Finalmente, es hora de ejecutar algunas consultas.

Número total de tanques informados (activos, inactivos)

from(bucket: "USTS") 
  |> range(start: -100y)
  |> filter(fn: (r) => r["_measurement"] == "fuel_tanks")
  |> filter(fn: (r) => r._field == "estimated_total_capacity")
  |> group(columns: ["status"])
  |> count(column: "_value")
  |> group(columns: ["_value", "status"], mode: "except")
  |> sort(columns: ["_value"], desc: true)

Los resultados en el gráfico a continuación muestran que hay 25,831 tanques de almacenamiento cerrados permanentemente en Connecticut en tres categorías: cierre permanente, uso actual y cierre temporal.

Número de tanques de combustible cerrados en todo momento por ciudad

Así que hay muchos tanques permanentemente cerrados. ¿Son sus distribuciones diferentes en el tiempo?

from(bucket: "USTS")
    |> range(start: -100y)
    |> filter(fn: (r) => r._measurement == "fuel_tanks" and r._field == "estimated_total_capacity" and r.status == "permanently closed")
    |> truncateTimeColumn(unit: 1y)
    |> group(columns: ["city", "_time"])
    |> count(column: "_value")
    |> drop(columns: ["closure_time", "construction_type", "overfill_protection", "substance_stored", "s2_cell_id_token", "lat", "lon"])
    |> group(columns: ["city"])

Esto generará un conjunto de tablas (pueblos) a lo largo del tiempo (usando Columna de tiempo truncado eliminar la granularidad de fecha y hora en los datos de la serie):

La figura muestra que el «ganador» tiene 80 tanques.

[ Learn how to manage your Linux environment for success. ]

Número de tanques, agrupados por tipo de sustancia

¿gasolina? ¿Aceite? Investigue lo que se usa actualmente en los últimos cinco años:

from(bucket: "USTS")
    |> range(start: -5y)
    |> filter(fn: (r) => r._measurement == "fuel_tanks" and r._field == "estimated_total_capacity" and r.status == "currently in use")
    |> group(columns: ["substance_stored"])
    |> count(column: "_value")
    |> drop(columns: ["city", "closure_tipe", "construction_type", "overfill_protection", "s2_cell_id", "lat", "lon", "_time", "spill_protection", "status"])
    |> group()
    |> sort(columns: ["_value"], desc: true)

resultado:

Los resultados muestran que la mayoría de los tanques de almacenamiento almacenan gasolina, seguido de aceite de calefacción para uso en sitio, diésel, tipos no específicos, queroseno para reventa, otro petróleo, aceite usado, combustible E15, aceite de calefacción para reventa, sustancias peligrosas, biodiésel, Combustible E85 y queroseno para uso in situ.

Número de tanques cerca de Hartford, Connecticut

Estoy usando Influx 2.4.0, que marcó el soporte nativo para la funcionalidad de geolocalización dentro de la base de datos como «experimental».pero es una característica muy útilasí que lo exploraré a continuación.

Si desea obtener una descripción general rápida de las capacidades de geolocalización, puede ejecutar los siguientes comandos en su computadora portátil:

import "influxdata/influxdb/sample"
import "experimental/geo"
sampleGeoData = sample.data(set: "birdMigration")
sampleGeoData
    |> geo.filterRows(region: {lat: 30.04, lon: 31.23, radius: 200.0}, strict: true)

Primero, un poco de geografía.Hartford, Connecticut, Estados Unidos, latitud y longitud Las coordenadas son 41.763710, -72.685097.

Calcule la cantidad de tanques dentro de un radio de 30 millas (48,28 km) de las coordenadas de Hartford en los últimos cinco años:

import "experimental/geo"
from(bucket: "USTS")
    |> range(start: -5y)
    |> filter(fn: (r) => r._measurement == "fuel_tanks" or r._field == "lat" or r.field == "lon" and r.status == "currently in use")
    |> geo.filterRows(region: {lat: 41.763710, lon: -72.685097, radius: 48.28032}, strict: false)
    |> drop(columns: ["closure_time", "construction_type", "overfill_protection", "substance_stored", "s2_cell_id_token", "lat", "lon", "spill_protection", "s2_cell_id", "_time", "status", "closure_type"])
    |> count(column: "estimated_total_capacity")
    |> group(columns: ["city"])
    |> group()
    |> sort(columns: ["estimated_total_capacity"], desc: true)

Aquí están algunos de los resultados:

Esto muestra la cantidad de tanques en las ciudades dentro de las 30 millas de Hartford; Hartford mismo tiene 173.

usar otros idiomas

También puede consultar los datos usando su lenguaje de programación favorito. Por ejemplo, aquí está la consulta de Python que uso para determinar cuántos tanques de cada tipo de sustancia están disponibles para un período determinado:

from influxdb_client import InfluxDBClient

# You can generate a Token from the "Tokens Tab" in the UI
token = "pP25Y9broJWTPfj_nPpSnGtFsoUtutOKsxP-AynRXJAz6fZzdhLCD4NqJC0eg_ImKDczbMQxMSTuhmsJHN7ikA=="
org = "Kodegeek"
bucket = "USTS"

with InfluxDBClient(url=" token=token, org=org) as client:
    query = """from(bucket: "USTS")
    |> range(start: -15y)
    |> filter(fn: (r) => r._measurement == "fuel_tanks" and r._field == "estimated_total_capacity" and r.status == "currently in use")
    |> group(columns: ["substance_stored"])
    |> count(column: "_value")
    |> drop(
        columns: [
            "city",
            "closure_type",
            "construction_type",
            "overfill_protection",
            "s2_cell_id",
            "lat",
            "lon",
            "_time",
            "spill_protection",
            "status",
        ],
    )
    |> group()
    |> sort(columns: ["_value"], desc: true)"""
    tables = client.query_api().query(query, org=org)
    for table in tables:
        for record in table.records:
            print(record)

Modificaré el código de Python y lo mejoraré:

No es tan malo. Puedes hacer lo mismo en otros lenguajes como Java.

[ Get the guide to installing applications on Linux. ]

¿Que sigue?

• Este Portal de datos abiertos de Connecticut Hay conjuntos de datos más interesantes que puede descargar de forma gratuita para aprender sobre el estado. Probablemente tengas un portal similar con datos públicos sobre dónde vives.
• Este Lenguaje de flujo Una opción más interesante que SQL simple para analizar datos de series temporales. Me resulta intuitivo de usar, pero todavía estoy aprendiendo algunos detalles.
• ¿Qué tan rápido es InfluxDB en comparación con otros productos? Tienes que decidir por ti mismo.no gasté Mucho tiempo ajustando la configuraciónMi interés inicial fue lo fácil que era de usar e integrar con otras aplicaciones como Grafana.
• Si está creando secuencias de comandos en Python, definitivamente debería echar un vistazo Ejemplo en el repositorio de Git.
• Y por supuesto, Revisa el código fuente en el repositorio de este tutorial.