Proceso Realtime Duoc

realtime.md

Instrucciones proceso Realtime

Previo a estos pasos se deberá tener acceso a la plataforma para generar datos realtime de DUOC https://bdrealtimeescuelait.duoc.cl/

Paso 1

Abrir laboratorio GSP823 - Preparing and Aggregating Data for Visualizations using Cloud Dataprep

Paso 2

Crear un topic en Pub/Sub con una id por ej: topic-ingesta y CREAR

Paso 3

Crear una Cloud Run Function, le ponemos un nombre por ejemplo "realtime-function"

Las opciones por defecto excepto

Región: La que de el lab

Autenticación: Requiere Autenticación

Ingress: All

container, volumes...

Request Timeout: 3060

Execution Environment: First Generation

CREATE

Paso 4

El cuerpo de la función será

from google.cloud import pubsub_v1
#import google.cloud.pubsub_v1 as pubsub_v1
import json
# Inicializa el cliente de Pub/Sub una vez
publisher = pubsub_v1.PublisherClient()
project_id = "[ID_DEL_PROYECTO]" # Reemplazar con tu ID de proyecto
topic_id = "topic-ingesta" # Reemplazar con topic_id elegido
topic_path = publisher.topic_path(project_id, topic_id)
def main(request):
    try:
        # Obtener el JSON desde el body del request
        data = request.get_json()
        if not data:
            return "Solicitud sin JSON válido", 400
        
        # Convertir a bytes para Pub/Sub
        message_bytes = json.dumps(data).encode("utf-8")
        # Publicar mensaje en Pub/Sub
        # Si es un array, publicar cada uno individualmente
        if isinstance(data, list):
            for item in data:
                message_bytes = json.dumps(item).encode("utf-8")
                future = publisher.publish(topic_path, message_bytes)
                future.result() # Esperar a que se complete la publicación
        else:
            message_bytes = json.dumps(data).encode("utf-8")
            future = publisher.publish(topic_path, message_bytes)
            future.result()

        return "Completado", 200
    except Exception as e:
        return f"Error al procesar la solicitud: {e}", 500

El punto de entrada será "main"

Requirements.txt con este cuerpo

google-cloud-pubsub
functions-framework
google-cloud-bigquery

Save and redeploy

Paso 5

Abrir Cloud Shell y darle permisos a la función con el siguiente comando

Reemplazar el nombre de la función "realtime-function" por el nombre de su función

gcloud run services add-iam-policy-binding realtime-function \
--region=us-central1 \
--member="allUsers" \
--role="roles/run.invoker"

Paso 6

Deshabilitar la Dataflow API y volver a habilitarla

Paso 7

Dar permisos en IAM Admin -> IAM

Agregar permisos de Pub/Sub a cuenta de servicio, se verá algo así "[email protected]"

Esta cuenta tiene el permiso de editor, haremos click en editar Y agregamos el Rol "Pub/Sub Publisher"

Apretamos "Save"

Paso 8

Volver al realtime function, entrar a la pestaña "Source" Hacer click en "Edit Source" y "Save and Redeploy" Esto para que vuelva a tomar los privilegios asociados a Pub/Sub y a los de la función

Paso 9

Ir a BigQuery -> Studio En nuestro nombre de proyectos, apretar el boton "..." y apretar "create database"

Nombre del dataset: DatosRealTime Seleccionar Multi-Region Región: US Crear Dataset

Paso 10

En nuestro dataset apretar boton "..." y apretar "create table"

Como nombre de tabla: DatosTR El schema seleccionar "Edit as Text" y usar el siguiente esquema:

[
    { "name": "id_cliente", "type": "STRING", "mode": "NULLABLE" },
    { "name": "cliente", "type": "STRING", "mode": "NULLABLE" },
    { "name": "genero", "type": "STRING", "mode": "NULLABLE" },
    { "name": "id_producto", "type": "STRING", "mode": "NULLABLE" },
    { "name": "producto", "type": "STRING", "mode": "NULLABLE" },
    { "name": "precio", "type": "FLOAT", "mode": "NULLABLE" },
    { "name": "cantidad", "type": "INTEGER", "mode": "NULLABLE" },
    { "name": "monto", "type": "FLOAT", "mode": "NULLABLE" },
    { "name": "forma_pago", "type": "STRING", "mode": "NULLABLE" },
    { "name": "fecreg", "type": "TIMESTAMP", "mode": "NULLABLE" }
]

Paso 11

Ir a Dataflow

Apretar "Create Job From Template" nombre: job-realtime-duoc Template: Pub/Sub Subscription to BigQuery

En Output Table: Seleccionamos la tabla de big Query Pub/Sub Subscription: Seleccionamos nuestro topic

En Streaming Engine, activar Streaming Engine

Temporary Location:

• Apretamos browse
• abrimos nuestro bucket
• creamos una carpeta tmp

Al finalizar apretamos run job

Paso 12

Vamos al cloud function y copiamos la URL de la función

vamos al sitio realtime de DUOC y lo pegamos ahí por ejemplo la URL https://cloud-run-function-realtime-178286789931.us-central1.run.app