sobre examinar las tendencias delictivas en su área local. Usted sabe que existen datos relevantes y tiene algunas habilidades analíticas básicas que puede utilizar para analizar estos datos. Sin embargo, estos datos cambian con frecuencia y usted desea mantener su análisis actualizado con los incidentes delictivos más recientes sin repetirlo. ¿Cómo podemos automatizar este proceso?
Bueno, si te has topado con este artículo, ¡estás de suerte! Juntos, veremos cómo crear un canal de datos para extraer datos de registros de la policía local y conectarlos a una plataforma de visualización para examinar las tendencias delictivas locales a lo largo del tiempo. Para este artículo, extraeremos datos sobre incidentes reportados al Departamento de Policía (CPD) de Cambridge (MA) y luego visualizaremos estos datos como un panel en Metabase.
Además, este artículo puede servir como plantilla general para cualquiera que busque escribir canalizaciones ETL orquestadas en Prefect y/o cualquiera que quiera conectar Metabase a sus almacenes de datos para crear análisis/informes detallados.
Nota: No tengo ninguna afiliación con Metabase; simplemente usaremos Metabase como plataforma de ejemplo para crear nuestro panel final. Hay muchas otras alternativas viables, que se describen en esta sección.
Contenido:
Conocimiento previo
Antes de profundizar en el proceso, será útil revisar los siguientes conceptos o mantener estos enlaces como referencia mientras lee.
Datos de Interés
Los datos con los que trabajaremos contienen una colección de entradas de registros policiales, donde cada entrada es un incidente único reportado al CPD. Cada entrada contiene información completa que describe el incidente, que incluye, entre otros:
Fecha y hora del incidente Tipo de incidente que ocurrió La calle donde ocurrió el incidente Una descripción en texto plano de lo que sucedió
Consulte el portal para obtener más información sobre los datos.
Para monitorear las tendencias delictivas en Cambridge, MA, es apropiado crear un canal de datos para extraer estos datos, ya que los datos se actualizan diariamente (según su sitio web). Si los datos se actualizaran con menos frecuencia (por ejemplo, anualmente), crear un canal de datos para automatizar este proceso no nos ahorraría mucho esfuerzo. Simplemente podríamos volver a visitar el portal de datos al final de cada año, descargar el archivo .csv y completar nuestro análisis.
Ahora que hemos encontrado el conjunto de datos adecuado, repasemos la implementación.
Canalización ETL
Para pasar de datos de registro de CPD sin procesar a un panel de metabase, nuestro proyecto constará de los siguientes pasos principales:
Extraiga los datos utilizando su API correspondiente. Transformarlo para prepararlo para el almacenamiento. Cargándolo en una base de datos PostgreSQL. Visualizando los datos en Metabase.
El flujo de datos de nuestro sistema se verá así:
Nuestro proceso sigue un flujo de trabajo ETL, lo que significa que transformaremos los datos antes de importarlos a PostgreSQL. Esto requiere cargar datos en la memoria mientras se ejecutan transformaciones de datos, lo que puede resultar problemático para conjuntos de datos grandes que son demasiado grandes para caber en la memoria. En este caso, podemos considerar un flujo de trabajo ELT, donde transformamos los datos en la misma infraestructura donde se almacenan. Dado que nuestro conjunto de datos es pequeño (<10k filas), esto no debería ser un problema y aprovecharemos el hecho de que pandas facilita la transformación de datos.
Extraeremos los datos del registro de CPD realizando una solicitud del conjunto de datos a la API de datos abiertos de Socrata. Usaremos sodapy, un cliente Python para la API, para realizar la solicitud.
Encapsularemos este código de extracción en su propio archivo: extract.py.
importar pandas como pd desde sodapy importar Socrata desde dotenv importar load_dotenv importar os desde prefecto importar tarea @task(retries=3, retry_delay_segundos=[10, 10, 10]) # reintentar solicitud de API en caso de falla def extract_data(): ''' Extraer datos de incidentes reportados al Departamento de Policía de Cambridge utilizando la API de datos abiertos de Socrata. Devuelve los datos del incidente como un Pandas DataFrame. ''' # recuperar el token de la aplicación Socrata desde .env # incluir este token de la aplicación al interactuar con la API de Socrata para evitar la limitación de solicitudes, de modo que podamos recuperar todos los incidentes load_dotenv() APP_TOKEN = os.getenv("SOCRATA_APP_TOKEN") # crear un cliente de Socrata para interactuar con la API de Socrata (https://github.com/afeld/sodapy) client = Socrata( "data.cambridgema.gov", APP_TOKEN, timeout=30 # aumentar el tiempo de espera desde el valor predeterminado de 10 segundos; a veces, lleva más tiempo obtener todos los resultados) # obtener todos los datos, paginando los resultados DATASET_ID = "3gki-wyrb" # identificador único para los datos del Registro de la Policía de Cambridge (https://data.cambridgema.gov/Public-Safety/Daily-Police-Log/3gki-wyrb/about_data) resultados = client.get_all(DATASET_ID) # Convertir a pandas DataFrame results_df = pd.DataFrame.from_records(resultados) return results_df
Notas sobre el código:
Socrata acelera las solicitudes si no incluye un token de aplicación que identifique de forma única su aplicación. Para obtener todos los resultados, incluiremos este token en nuestra solicitud y lo colocaremos en un archivo .env para mantenerlo fuera de nuestro código fuente. Especificaremos un tiempo de espera de 30 segundos (en lugar del tiempo de espera predeterminado de 10 segundos) al realizar nuestra solicitud a la API de Socrata. Según la experiencia con el uso de la API, obtener todos los resultados a veces podía llevar más de 10 segundos y, por lo general, 30 segundos eran suficientes para evitar errores de tiempo de espera. Cargaremos los resultados obtenidos en un DataFrame de pandas, ya que validaremos y transformaremos estos datos usando pandas.
ETL: validar
Ahora, haremos algunas comprobaciones básicas de calidad de los datos.
Los datos ya están bastante limpios (lo cual tiene sentido ya que los proporciona el Departamento de Policía de Cambridge). Por lo tanto, nuestros controles de calidad de datos actuarán más como un "control de cordura" para comprobar que no hemos ingerido nada inesperado.
Validaremos lo siguiente:
Todas las columnas esperadas (como se especifica aquí) están presentes. Todas las identificaciones son numéricas. Las fechas y horas siguen el formato ISO 8601. No faltan valores en las columnas que deberían contener datos. Específicamente, cada incidente debe tener una fecha, hora, ID, tipo y ubicación.
Pondremos este código de validación en su propio archivo: validar.py.
desde fecha y hora importar fecha y hora desde colecciones importar contador importar pandas como pd desde tarea de importación perfecta ### UTILIDADES def check_valid_schema(df): ''' Compruebe si el contenido del DataFrame contiene las columnas esperadas para el conjunto de datos de la policía de Cambridge. De lo contrario, genera un error. ''' SCHEMA_COLS = ['date_time', 'id', 'type', 'subtype', 'ubicación', 'last_updated', 'description'] if Counter(df.columns) != Counter(SCHEMA_COLS): rise ValueError("El esquema no coincide con el esquema esperado.") def check_numeric_id(df): ''' Convertir valores de 'id' a numéricos. Si algún valor de 'id' no es numérico, reemplácelo con NaN, para que pueda eliminarse posteriormente en las transformaciones de datos. ''' df['id'] = pd.to_numeric(df['id'], errores='coerce') return df def verificar_datetime(df): ''' Verifique que los valores de 'date_time' sigan el formato ISO 8601 (https://www.iso.org/iso-8601-date-and-time-format.html). Genera un ValueError si alguno de los valores de 'fecha_hora' no es válido. ''' df.apply(lambda fila: datetime.fromisoformat(row['date_time']), axis=1) def check_missing_values(df): ''' Compruebe si faltan valores en las columnas que requieren datos. Para los registros policiales, cada incidente debe tener una fecha y hora, ID, tipo de incidente y ubicación. ''' REQUIRED_COLS = ['date_time', 'id', 'type', 'location'] para col en REQUIRED_COLS: if df[col].isnull().sum() > 0: rise ValueError(f"Hay valores faltantes en el atributo '{col}'.") ### LÓGICA DE VALIDACIÓN @task def validar_data(df): ''' Verifique que los datos cumplan con los siguientes controles de calidad de datos: – el esquema es válido – Los ID son numéricos – la fecha y hora sigue el formato ISO 8601 – no faltan valores en las columnas que requieren datos ''' check_valid_schema(df) df = check_numeric_id(df) verificar_datetime(df) check_missing_values(df) return df
Al implementar estas comprobaciones de calidad de los datos, es importante pensar en cómo manejar las comprobaciones de calidad de los datos que fallan.
¿Queremos que nuestra canalización falle estrepitosamente (por ejemplo, genere un error/bloqueo)? ¿Debería nuestro oleoducto manejar las fallas en silencio? Por ejemplo, ¿marcar los datos identificados como no válidos para que puedan eliminarse posteriormente?
Generaremos un error si:
Los datos ingeridos no siguen el esquema esperado. No tiene sentido procesar los datos si no contienen lo que esperamos. La fecha y hora no sigue el formato ISO 8601. No existe una forma estándar de convertir valores de fecha y hora incorrectos a su formato de fecha y hora correcto correspondiente. El incidente contiene valores faltantes para cualquiera de fecha, hora, ID, tipo y ubicación. Sin estos valores, el incidente no se puede describir de manera integral.
Para los registros que tienen ID no numéricos, los rellenaremos con marcadores de posición NaN y luego los eliminaremos en el paso de transformación. Estos registros no interrumpen nuestro análisis si simplemente los eliminamos.
ETL: transformar
Ahora, haremos algunas transformaciones en nuestros datos para prepararlos para el almacenamiento en PostgreSQL.
Haremos las siguientes transformaciones:
Elimine filas duplicadas: usaremos la columna 'ID' para identificar duplicados. Elimine filas no válidas: algunas de las filas que no pasaron las comprobaciones de calidad de los datos se marcaron con un 'ID' de NaN, por lo que las eliminaremos. Divida la columna de fecha y hora en columnas separadas de año, mes, día y hora. En nuestro análisis final, es posible que deseemos analizar las tendencias delictivas en estos diferentes intervalos de tiempo, por lo que crearemos estas columnas adicionales aquí para simplificar nuestras consultas posteriores.
Pondremos este código de transformación en su propio archivo: transform.py.
importar pandas como pd desde la tarea de importación perfecta ### UTILIDADES def remove_duplicates(df): ''' Eliminar filas duplicadas del marco de datos según la columna 'id'. Mantenga la primera aparición. ''' return df.drop_duplicates(subset=["id"], keep='first') def remove_invalid_rows(df): ''' Elimina las filas donde el 'id' es NaN, ya que estos ID se identificaron como no numéricos. ''' return df.dropna(subset='id') def split_datetime(df): ''' Divida la columna date_time en columnas separadas de año, mes, día y hora. ''' # convertir a fecha y hora df['date_time'] = pd.to_datetime(df['date_time']) # extraer año/mes/día/hora df['year'] = df['date_time'].dt.year df['month'] = df['date_time'].dt.month df['day'] = df['date_time'].dt.day df['hour'] = df['date_time'].dt.hour df['minuto'] = df['date_time'].dt.minuto df['segundo'] = df['date_time'].dt.segundo return df ### LÓGICA DE TRANSFORMACIÓN @task def transform_data(df): ''' Aplique las siguientes transformaciones al marco de datos pasado: – deduplicar registros (conserve el primero) – eliminar filas no válidas – dividir fecha y hora en año, columnas de mes, día y hora ''' df = remove_duplicates(df) df = remove_invalid_rows(df) df = split_datetime(df) return df
ETL: cargar
Ahora nuestros datos están listos para importar a PostgreSQL.
Antes de que podamos importar nuestros datos, necesitamos crear nuestra instancia de PostgreSQL. Crearemos uno localmente usando un archivo de redacción. Este archivo nos permite definir y configurar todos los servicios que nuestra aplicación necesita.
servicios: postgres_cpd: # instancia de postgres para la imagen de canalización ETL de CPD: postgres:16 nombre_contenedor: postgres_cpd_dev entorno: POSTGRES_USER: ${POSTGRES_USER} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} POSTGRES_DB: cpd_db puertos: – "5433:5432" # Postgres ya está en el puerto 5432 en mi volúmenes de máquina local: – pgdata_cpd:/var/lib/postgresql/data restart: a menos que se detenga pgadmin: imagen: dpage/pgadmin4 nombre_contenedor: pgadmin_dev entorno: PGADMIN_DEFAULT_EMAIL: ${PGADMIN_DEFAULT_EMAIL} PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_DEFAULT_PASSWORD} puertos: – "8081:80" depende_on: # no inicie pg_admin hasta que nuestra instancia de postgres se esté ejecutando – volúmenes postgres_cpd: pgdata_cpd: # todos los datos de nuestro servicio postgres_cpd se almacenarán aquí
Hay dos servicios principales definidos aquí:
postgres_cpd: esta es nuestra instancia de PostgreSQL donde almacenaremos nuestros datos. pgadmin: plataforma de administración de base de datos que proporciona una GUI que podemos usar para consultar datos en nuestra base de datos PostgreSQL. No es funcionalmente necesario, pero es útil para comprobar los datos de nuestra base de datos. Para obtener más información sobre cómo conectarse a su base de datos PostgreSQL en pgAdmin, haga clic aquí.
Resaltemos algunas configuraciones importantes para nuestro servicio postgres_cpd:
nombre_contenedor: postgres_cpd_dev -> Nuestro servicio se ejecutará en un contenedor (es decir, un proceso aislado) llamado postgres_cpd_dev. Docker genera nombres de contenedores aleatorios si no lo especifica, por lo que asignar un nombre hará que sea más sencillo interactuar con el contenedor. entorno: -> Creamos un usuario de Postgres a partir de las credenciales almacenadas en nuestro archivo .env. Además, creamos una base de datos predeterminada, cpd_dev. puertos: -> Nuestro servicio PostgreSQL escuchará en el puerto 5432 dentro del contenedor. Sin embargo, asignaremos el puerto 5433 en la máquina host al puerto 5432 en el contenedor, lo que nos permitirá conectarnos a PostgreSQL desde nuestra máquina host a través del puerto 5433. volúmenes: -> Nuestro servicio almacenará todos sus datos (por ejemplo, configuración, archivos de datos) en el siguiente directorio dentro del contenedor: /var/lib/postgresql/data. Montaremos este directorio de contenedor en un volumen Docker con nombre almacenado en nuestra máquina local, pgdata_cpd. Esto nos permite conservar los datos de la base de datos más allá de la vida útil del contenedor.
Ahora que hemos creado nuestra instancia de PostgreSQL, podemos ejecutar consultas en ella. Importar nuestros datos a PostgreSQL requiere ejecutar dos consultas en la base de datos:
Creando la tabla que almacenará los datos. Cargando nuestros datos transformados en esa tabla.
Cada vez que ejecutamos una consulta en nuestra instancia de PostgreSQL, debemos hacer lo siguiente:
Establecer nuestra conexión a PostgreSQL. Ejecute la consulta. Confirme los cambios y cierre la conexión. from prefect import task from sqlalchemy import create_engine import psycopg2 from dotenv import load_dotenv import os # leer contenido de .env, que contiene nuestras credenciales de Postgres load_dotenv() def create_postgres_table(): ''' Cree la tabla cpd_incidents en Postgres DB (cpd_db) si no existe. ''' # establecer conexión con la base de datos conn = psycopg2.connect( host="localhost", port="5433", base de datos="cpd_db", usuario=os.getenv("POSTGRES_USER"), contraseña=os.getenv("POSTGRES_PASSWORD") ) # crear objeto cursor para ejecutar SQL cur = conn.cursor() # ejecutar consulta para crear la tabla create_table_query = ''' CREAR TABLA SI NO EXISTE cpd_incidents ( fecha_hora TIMESTAMP, id INTEGER PRIMARY KEY, tipo TEXTO, subtipo TEXTO, ubicación TEXTO, descripción TEXTO, última_actualización TIMESTAMP, año INTEGER, mes INTEGER, día INTEGER, hora INTEGER, minuto INTEGER, segundo INTEGER ) ''' cur.execute(create_table_query) # confirmar cambios conn.commit() # cerrar cursor y conexión cur.close() conn.close() @task def load_into_postgres(df): ''' Carga los datos transformados pasados como un DataFrame en la tabla 'cpd_incidents' en nuestra instancia de Postgres. ''' # crear una tabla para insertar datos según sea necesario create_postgres_table() # crear un objeto de motor para conectarse al motor de base de datos = create_engine(f"postgresql://{os.getenv("POSTGRES_USER")}:{os.getenv("POSTGRES_PASSWORD")}@localhost:5433/cpd_db") # insertar datos en la base de datos de Postgres en la tabla 'cpd_incidents' df.to_sql('cpd_incidents', motor, if_exists='reemplazar')
Cosas a tener en cuenta sobre el código anterior:
De manera similar a cómo obtuvimos el token de nuestra aplicación para extraer nuestros datos, recuperaremos nuestras credenciales de Postgres de un archivo .env. Para cargar el DataFrame que contiene nuestros datos transformados en Postgres, usaremos pandas.DataFrame.to_sql(). Es una forma sencilla de insertar datos de DataFrame en cualquier base de datos compatible con SQLAlchemy.
Definición del canal de datos
Hemos implementado los componentes individuales del proceso ETL. Ahora estamos listos para encapsular estos componentes en una canalización.
Hay muchas herramientas disponibles para orquestar canalizaciones definidas en Python. Dos opciones populares son Apache Airflow y Prefect.
Para simplificar, procederemos a definir nuestra canalización utilizando Prefect. Necesitamos hacer lo siguiente para comenzar:
Instale Prefect en nuestro entorno de desarrollo. Obtenga un servidor API perfecto. Como no queremos administrar nuestra propia infraestructura para ejecutar Prefect, nos registraremos en Prefect Cloud.
Para obtener más información sobre la configuración de Prefect, consulte los documentos.
A continuación, debemos agregar los siguientes decoradores a nuestro código:
@task -> Agregue esto a cada función que implemente un componente de nuestra canalización ETL (es decir, nuestras funciones de extracción, validación, transformación y carga). @flow -> Agregue este decorador a la función que encapsula los componentes ETL en una canalización ejecutable.
Si vuelve a mirar nuestro código de extracción, validación, transformación y carga, verá que agregamos el decorador @task a estas funciones.
Ahora, definamos nuestra canalización ETL que ejecuta estas tareas. Pondremos lo siguiente en un archivo separado, etl_pipeline.py.
de extraer importar extraer_datos de validar importar validar_datos de transformar importar transformar_datos de cargar importar load_into_postgres de prefecto importar flujo @flow(name="cpd_incident_etl", log_prints=True) # Nuestra canalización aparecerá como 'cpd_incident_etl' en la interfaz de usuario de Prefecto. Todos los resultados de impresión se mostrarán en Prefect. def etl(): ''' Ejecutar el proceso ETL: – Extraer datos de incidentes de CPD de la API de Socrata – Validar y transformar los datos extraídos para prepararlos para el almacenamiento – Importar los datos transformados a Postgres ''' print("Extrayendo datos…") extraído_df = extract_data() print("Realizando controles de calidad de datos…") valided_df = validar_data(extracted_df) print("Realizando transformaciones de datos…") transform_df = transform_data(validated_df) print("Importando datos a Postgres…") load_into_postgres(transformed_df) print("ETL complete!") if __name__ == "__main__": # Se espera que los datos de CPD se actualicen diariamente (https://data.cambridgema.gov/Public-Safety/Daily-Police-Log/3gki-wyrb/about_data) # Por lo tanto, ejecutaremos nuestra canalización en un diariamente (a medianoche) etl.serve(name="cpd-pipeline-deployment", cron="0 0 * * *")
Cosas a tener en cuenta sobre el código:
@flow(name=”cpd_incident_etl”, log_prints=True) -> esto nombra nuestra canalización “cpd_incident_etl”, que se reflejará en la interfaz de usuario de Prefect. La salida de todas nuestras declaraciones impresas se registrará en Prefect. etl.serve(name=”cpd-pipeline-deployment”, cron=”0 0 * * *”) -> esto crea una implementación de nuestra canalización, denominada “cpd-pipeline-deployment”, que se ejecuta todos los días a medianoche.
Ahora que hemos creado nuestra canalización para cargar nuestros datos en PostgreSQL, es hora de visualizarla.
Hay muchos enfoques que podríamos adoptar para visualizar nuestros datos. Algunas opciones notables incluyen:
Ambas son buenas opciones. Sin entrar en demasiados detalles detrás de cada herramienta de BI, usaremos Metabase para crear un panel simple.
Metabase es una herramienta de análisis integrada y BI de código abierto que simplifica la visualización y el análisis de datos. Conectar Metabase a nuestras fuentes de datos e implementarlo es sencillo, en comparación con otras herramientas de BI (por ejemplo, Apache Superset).
En el futuro, si queremos tener una mayor personalización de nuestros elementos visuales/informes, podemos considerar el uso de otras herramientas. Por ahora, Metabase servirá para crear una prueba de concepto.
Metabase le permite elegir entre usar su versión en la nube o administrar una instancia autohospedada. Metabase Cloud ofrece varios planes de pago, pero puedes crear una instancia autohospedada de Metabase de forma gratuita utilizando Docker. Definiremos nuestra instancia de Metabase en nuestro archivo de redacción.
Dado que somos autohospedados, también tenemos que definir la base de datos de la aplicación Metabase, que contiene los metadatos que Metabase necesita para consultar sus fuentes de datos (en nuestro caso, postgres_cpd). servicios: postgres_cpd: # instancia de postgres para la imagen de canalización ETL de CPD: postgres:16 nombre_contenedor: postgres_cpd_dev entorno: POSTGRES_USER: ${POSTGRES_USER} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} POSTGRES_DB: cpd_db puertos: – "5433:5432" # Postgres ya está en el puerto 5432 en mi volúmenes de máquinas locales: – pgdata_cpd:/var/lib/postgresql/data restart: redes a menos que se detengan: – metanet1 pgadmin: imagen: dpage/pgadmin4 nombre_contenedor: pgadmin_dev entorno: PGADMIN_DEFAULT_EMAIL: ${PGADMIN_DEFAULT_EMAIL} PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_DEFAULT_PASSWORD} puertos: – "8081:80" depende_de: – redes postgres_cpd: – metabase metanet1: # tomado de https://www.metabase.com/docs/latest/installation-and-operation/running-metabase-on-docker imagen: metabase/metabase: último nombre_contenedor: nombre de host de la metabase: volúmenes de la metabase: – /dev/urandom:/dev/random:ro ports: – Entorno "3000:3000": MB_DB_TYPE: postgres MB_DB_DBNAME: metabaseappdb MB_DB_PORT: 5432 MB_DB_USER: ${METABASE_DB_USER} MB_DB_PASS: ${METABASE_DB_PASSWORD} MB_DB_HOST: postgres_metabase # debe coincidir con el nombre del contenedor de redes postgres_mb (instancia de Metabase Postgres): – chequeo de salud metanet1: prueba: curl –fail -I http://localhost:3000/api/health || intervalo de salida 1: 15 s tiempo de espera: 5 s reintentos: 5 postgres_mb: # instancia de postgres para administrar la imagen de la instancia de Metabase: postgres:16 nombre_contenedor: postgres_metabase # otros servicios deben usar este nombre para comunicarse con este contenedor nombre de host: postgres_metabase # identificador interno, no afecta la comunicación con otros servicios (útil para registros) entorno: POSTGRES_USER: ${METABASE_DB_USER} POSTGRES_DB: metabaseappdb POSTGRES_PASSWORD: ${METABASE_DB_PASSWORD} puertos: – volúmenes "5434:5432": – pgdata_mb:/var/lib/postgresql/data redes: – metanet1 # Aquí, definiremos volúmenes separados para aislar la configuración de base de datos y los archivos de datos para cada base de datos de Postgres. # Nuestra base de datos Postgres para nuestra aplicación debe almacenar su configuración/datos por separado de la base de datos Postgres en la que se basa nuestro servicio Metabase. volúmenes: pgdata_cpd: pgdata_mb: # define la red sobre la cual se comunicarán todos los servicios redes: metanet1: driver: bridge # PARA HACER: 'bridge' es la red predeterminada – los servicios podrán comunicarse entre sí usando sus nombres de servicio
Para crear nuestra instancia de Metabase, realizamos los siguientes cambios en nuestro archivo de redacción:
Se agregaron dos servicios: metabase (nuestra instancia de Metabase) y postgres_mb (la base de datos de la aplicación de nuestra instancia de Metabase). Se definió un volumen adicional, pgdata_mb. Esto almacenará los datos para la base de datos de la aplicación Metabase (postgres_mb). Definida la red sobre la cual se comunicarán los servicios, metanet1.
Sin entrar en demasiados detalles, analicemos los servicios de metabase y postgres_mb.
Nuestra instancia de Metabase (metabase):
Este servicio estará expuesto en el puerto 3000 de la máquina host y dentro del contenedor. Si ejecutamos este servicio en nuestra máquina local, podremos acceder a él en localhost:3000. Conectamos Metabase a la base de datos de su aplicación asegurándonos de que las variables de entorno MB_DB_HOST, MB_DB_PORT y MB_DB_NAME coincidan con el nombre del contenedor, los puertos y el nombre de la base de datos que figuran en el servicio postgres_mb.
Para obtener más información sobre cómo ejecutar Metabase en Docker, consulte los documentos.
Después de configurar Metabase, se le pedirá que conecte Metabase a su fuente de datos.
Después de seleccionar una fuente de datos de PostgreSQL, podemos especificar la siguiente cadena de conexión para conectar Metabase a nuestra instancia de PostgreSQL, sustituyendo sus credenciales según sea necesario:
postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@postgres_cpd:5432/cpd_db
Después de configurar la conexión, podemos crear nuestro panel de control. Puede crear una amplia variedad de elementos visuales en Metabase, por lo que no entraremos en detalles aquí.
Revisemos el panel de ejemplo que mostramos al principio de esta publicación. Este panel resume muy bien las tendencias recientes e históricas en los incidentes de CPD reportados.
Desde este panel podemos ver lo siguiente:
La mayoría de los incidentes se informan al CPD a media tarde. Una abrumadora mayoría de los incidentes denunciados son del tipo “INCIDENTE”. El número de incidentes reportados alcanzó su punto máximo entre agosto y octubre de 2025 y ha ido disminuyendo constantemente desde entonces.
Afortunadamente para nosotros, Metabase consultará nuestra base de datos cada vez que carguemos este panel, por lo que no tendremos que preocuparnos de que este panel muestre datos obsoletos.
Consulte el repositorio de Git aquí si desea profundizar en la implementación.
Resumen y trabajo futuro
¡Gracias por leer! Recapitulemos brevemente lo que construimos:
Creamos una canalización de datos para extraer, transformar y cargar datos de Cambridge Police Log en una base de datos PostgreSQL autohospedada. Implementamos esta canalización utilizando Prefect y la programamos para que se ejecutara diariamente. Creamos una instancia autohospedada de Metabase, la conectamos a nuestra base de datos PostgreSQL y creamos un panel para visualizar las tendencias delictivas recientes e históricas en Cambridge, MA.
Hay muchas maneras de aprovechar este proyecto, incluidas, entre otras:
Crear visualizaciones adicionales (mapa de calor geoespacial) para visualizar la frecuencia de los delitos en diferentes áreas dentro de Cambridge. Esto requeriría transformar nuestros datos de ubicación de calles en coordenadas de latitud/longitud. Implementar nuestra canalización y servicios autohospedados fuera de nuestra máquina local. Considere unir estos datos con otros conjuntos de datos para realizar análisis profundos entre dominios. Por ejemplo, tal vez podríamos unir este conjunto de datos a datos demográficos/censales (usando la ubicación de la calle) para ver si las áreas de diferente composición demográfica dentro de Cambridge tienen diferentes tasas de incidentes.
Si tienes alguna otra idea sobre cómo ampliar este proyecto, o si hubieras construido las cosas de manera diferente, ¡me encantaría escucharla en los comentarios!
El autor ha creado todas las imágenes de este artículo.
Fuentes y GitHub
Prefecto:
Metabase:
Estibador:
Repositorio de GitHub:
Conjunto de datos de registro policial diario de CPD: