Transfiere datos CSV (valores separados por comas) a BigQuery con Cloud Data Fusion: transferencia en tiempo real

1. Introducción

Última actualización: 28 de febrero de 2020

En este codelab, se muestra un patrón de transferencia de datos para transferir datos de atención médica con formato CSV a BigQuery en tiempo real. Para este lab, usaremos la canalización de datos en tiempo real de Cloud Data Fusion. Se generaron datos realistas de pruebas de salud y se pusieron a disposición en el bucket de Google Cloud Storage (gs://hcls_testing_data_fhir_10_patients/csv/).

En este codelab, aprenderás lo siguiente:

• Cómo transferir datos de CSV (carga en tiempo real) de Pub/Sub a BigQuery con Cloud Data Fusion.
• Cómo crear de forma visual una canalización de integración de datos en Cloud Data Fusion para cargar, transformar y enmascarar datos de atención médica en tiempo real

¿Qué necesitas para ejecutar esta demostración?

• Necesitas acceso a un proyecto de GCP.
• Debes tener asignada una función de propietario para el proyecto de GCP.
• Datos de atención médica en formato CSV, incluido el encabezado.

Si no tienes un proyecto de GCP, sigue estos pasos para crear uno nuevo.

Los datos de atención médica en formato CSV se cargaron previamente en el bucket de GCS, en gs://hcls_testing_data_fhir_10_patients/csv/. Cada archivo de recursos CSV tiene una estructura de esquema única. Por ejemplo, Patients.csv tiene un esquema diferente al de Providers.csv. Los archivos de esquema precargados se pueden encontrar en gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Si necesitas un conjunto de datos nuevo, siempre puedes generarlo con SyntheaTM. Luego, súbelos a GCS en lugar de copiarlos del bucket en el paso Copiar datos de entrada.

Referencia:

Operaciones básicas en Cloud Data Fusion (guía de inicio rápido) con Google Cloud Platform Console

Mensajes de publicación de Cloud Pub/Sub https://cloud.google.com/pubsub/docs/publisher.

2. Configuración de un proyecto de GCP

Inicializar variables de shell para tu entorno

Para encontrar el PROJECT_ID, consulta Identifica proyectos.

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

Crea un bucket de GCS para almacenar datos de entrada y registros de errores con la herramienta gsutil.

gsutil mb -l us gs://$BUCKET_NAME

Obtén acceso al conjunto de datos sintético.

1. Desde la dirección de correo electrónico que usas para acceder a la consola de Cloud, envía un correo electrónico a [e-mail] para solicitar unirte.
2. Recibirás un correo electrónico con instrucciones para confirmar la acción.
3. Usa la opción de responder el correo electrónico para unirte al grupo. NO hagas clic en el botón .
4. Una vez que recibas el correo electrónico de confirmación, puedes continuar con el siguiente paso del codelab.

Copia los datos de entrada.

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

Crear un conjunto de datos de BigQuery

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

Recuerda: Para este lab, siempre puedes generar un nuevo conjunto de datos de prueba realistas con SyntheaTM. Para tu comodidad, te ofrecemos un conjunto de datos en el siguiente almacenamiento de Google Cloud. gs://hcls_testing_data_fhir_10_patients/csv.

Instala e inicializa el SDK de Google Cloud y crea el tema y las suscripciones de Pub o Sub.

gcloud init
gcloud pubsub topics create your-topic
gcloud pubsub subscriptions create --topic your-topic your-sub

3. Configuración del entorno de Cloud Data Fusion

Sigue estos pasos para habilitar la API de Cloud Data Fusion y otorgar los permisos necesarios:

Habilita las API.

1. Ve a la Biblioteca de API de GCP Console.
2. Selecciona el proyecto desde la lista de proyectos.
3. En la Biblioteca de APIs, selecciona la API que deseas habilitar ( API de Cloud Data Fusion,API de Cloud Pub/Sub). Si necesitas ayuda para encontrar la API, usa el campo de búsqueda y los filtros.
4. En la página de API, haz clic en HABILITAR.

Crea una instancia de Cloud Data Fusion.

1. En GCP Console, selecciona el ID del proyecto.
2. Selecciona Data Fusion en el menú de la izquierda y, luego, haz clic en el botón CREATE AN INSTANCE en el medio de la página (primera creación) o haz clic en el botón CREATE INSTANCE en el menú superior (creación adicional).

3. Proporciona el nombre de la instancia. Selecciona Enterprise.

4. Haz clic en el botón CREAR.

Nota: El proceso de creación de la instancia puede tardar hasta 30 minutos.

Configura permisos de instancia.

Después de crear una instancia, sigue estos pasos para otorgar a la cuenta de servicio asociada con la instancia los permisos para tu proyecto:

Nota: La concesión de permisos debe realizarse solo una vez en un proyecto.

1. Haga clic en el nombre de la instancia para navegar a su página de detalles.

2. Copia la cuenta de servicio.

3. Navega a la página IAM de tu proyecto.
4. En la página de permisos de IAM, haz clic en el botón Agregar para otorgar a la cuenta de servicio el rol Agente de servicio de la API de Cloud Data Fusion. Pega la "cuenta de servicio" en el campo Miembros nuevos y selecciona Administración de servicio -> Rol de agente del servidor de la API de Cloud Data Fusion.

5. Haz clic en + Agregar otro rol (o Editar agente de servicio de la API de Cloud Data Fusion) para agregar un rol de suscriptor de Pub/Sub.

6. Haz clic en Guardar.

Una vez que hayas completado estos pasos, puedes comenzar a usar Cloud Data Fusion haciendo clic en el vínculo Ver instancia en la página de instancias de Cloud Data Fusion o en la página de detalles de una instancia.

Configura la regla de firewall.

1. Navega a la consola de GCP -> Red de VPC -> Reglas de firewall para verificar si existe o no la regla default-allow-ssh.

2. De lo contrario, agrega una regla de firewall que permita todo el tráfico SSH de entrada a la red predeterminada.

Con la línea de comandos:

gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging

Mediante la IU: Haz clic en Crear regla de firewall y completa la información:

4. Compila nodos para la canalización

Ahora que tenemos el entorno de Cloud Data Fusion en GCP, siga estos pasos para comenzar a compilar las canalizaciones de datos en Cloud Data Fusion:

1. En la ventana de Cloud Data Fusion, haz clic en el vínculo Ver instancia en la columna Acción. Se te redireccionará a otra página. Haz clic en la url proporcionada para abrir la instancia de Cloud Data Fusion. Hacer clic en "Iniciar visita guiada" o "No, gracias" de la ventana emergente de bienvenida.
2. Expandir el menú “hamburguesa” en el menú, selecciona Canalización -> Mostrar en lista

3. Haz clic en el botón verde + en la esquina superior derecha y, luego, selecciona Crear canalización. O haz clic en "Crear" un vínculo de canalización.

4. Cuando aparezca el estudio de canalización, en la parte superior izquierda, selecciona Data Pipeline - Realtime en el menú desplegable.

5. En la IU de Data Pipelines, verás diferentes secciones en el panel izquierdo como Filtro, Fuente, Transformación, Estadísticas, Receptor, Controladores de errores y alertas, en las que podrás seleccionar uno o varios nodos para la canalización.

Selecciona un nodo de origen.

1. En la sección Fuente en la paleta de complementos a la izquierda, haz doble clic en el nodo de Google Cloud PubSub, que aparece en la IU de Data Pipelines.
2. Dirígete al nodo fuente de Pub/Sub y haz clic en Propiedades.

3. Llena los campos obligatorios. Configura los siguientes campos:

• Etiqueta = {any text}
• Nombre de referencia = {any text}
• ID del proyecto = detección automática
• Suscripción: Suscripción creada en la sección Crear tema de Pub/Sub (por ejemplo, tu-suscriptor)
• Tema = Tema creado en la sección Crear tema de Pub/Sub (por ejemplo, tu-tema)

4. Haz clic en Documentation para obtener una explicación detallada. Haz clic en el botón Validar (Validate) para validar toda la información de entrada. Verde "No se encontraron errores" indica éxito.

5. Para cerrar las propiedades de Pub/Sub, haz clic en el botón X.

Selecciona el nodo Transform.

1. En la sección Transform en la paleta de complementos a la izquierda, haz doble clic en el nodo Projection, que aparece en la IU de Data Pipelines. Conecta el nodo fuente de Pub/Sub al nodo de transformación de proyección.
2. Selecciona el nodo Projection y haz clic en Properties.

3. Llena los campos obligatorios. Configura los siguientes campos:

• Convert = convierte message de tipo de byte a tipo de string.
• Campos que se van a descartar = {any field}
• Campos que se conservarán = {message, timestamp y attributes} (por ejemplo, atributos: key=‘filename':value=‘patients’ enviado desde Pub/Sub)
• Campos para cambiar el nombre = {message, timestamp}

4. Haz clic en Documentation para obtener una explicación detallada. Haz clic en el botón Validar (Validate) para validar toda la información de entrada. Verde "No se encontraron errores" indica éxito.

5. En la sección Transformar en la paleta de complementos a la izquierda, haz doble clic en el nodo Wrangler, que aparece en la IU de Data Pipelines. Conecta el nodo de transformación de proyección a un nodo de transformación de Wrangler. Apunta al nodo Wrangler y haz clic en Properties.

6. Haz clic en el menú desplegable Acciones y selecciona Importar para importar un esquema guardado (por ejemplo: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ esquema (Patients.json).
7. Agrega el campo TIMESTAMP en el esquema de salida (si no existe). Para ello, haz clic en el botón + junto al último campo y marca “Nulo”. .
8. Llena los campos obligatorios. Configura los siguientes campos:

• Etiqueta = {any text}
• Nombre del campo de entrada = {*}
• Precondition = {attributes.get("filename") != "patients"} para distinguir cada tipo de registro o mensaje (por ejemplo, pacientes, proveedores, alergias, etc.) que se envía desde el nodo fuente de Pub/Sub.

9. Haz clic en Documentation para obtener una explicación detallada. Haz clic en el botón Validar (Validate) para validar toda la información de entrada. Verde "No se encontraron errores" indica éxito.

10. Configura los nombres de las columnas en un orden preferido y descarta los campos que no necesites. Copia el siguiente fragmento de código y pégalo en el cuadro Receta.

drop attributes
parse-as-csv :body ',' false
drop body
set columns TIMESTAMP,Id,BIRTHDATE,DEATHDATE,SSN,DRIVERS,PASSPORT,PREFIX,FIRST,LAST,SUFFIX,MAIDEN,MARITAL,RACE,ETHNICITY,GENDER,BIRTHPLACE,ADDRESS,CITY,STATE,ZIP
mask-number SSN xxxxxxx####

Nota: Falso significa analizar la columna del cuerpo con la opción "Set first row as header". opción no marcada.

11. Consulta Codelab por lotes: CSV a BigQuery a través de CDF para enmascarar y desidentificación de datos. O agrega este fragmento de código mask-number SSN xxxxxxx#### en el cuadro Receta.
12. Para cerrar la ventana Transform Properties, haz clic en el botón X.

Selecciona el nodo Receptor.

1. En la sección Receptor en la paleta de complementos a la izquierda, haz doble clic en el nodo de BigQuery, que aparece en la IU de Data Pipeline. Conectar el nodo de transformación de Wrangler al nodo receptor de BigQuery
2. Coloca el cursor sobre el nodo receptor de BigQuery y haz clic en Propiedades.

3. Completa los campos obligatorios:

• Etiqueta = {any text}
• Nombre de referencia = {any text}
• ID del proyecto = detección automática
• Conjunto de datos = Conjunto de datos de BigQuery que se usa en el proyecto actual (por ejemplo, DATASET_ID)
• Tabla = {table name}

4. Haz clic en Documentation para obtener una explicación detallada. Haz clic en el botón Validar (Validate) para validar toda la información de entrada. Verde "No se encontraron errores" indica éxito.

5. Para cerrar las Propiedades de BigQuery, haz clic en el botón X.

5. Compila una canalización de datos en tiempo real

En la sección anterior, creamos nodos necesarios para crear una canalización de datos en Cloud Data Fusion. En esta sección, conectaremos los nodos para compilar la canalización real.

Conecta todos los nodos de una canalización

1. Arrastra una flecha de conexión > en el borde derecho del nodo de origen y soltarlo en el borde izquierdo del nodo de destino.
2. Una canalización puede tener varias ramas que obtienen mensajes publicados del mismo nodo fuente de Pub/Sub.

Nota: Compila una rama desde el nodo de proyección en la canalización para cada mensaje de Pub/Sub.

3. Asigna un nombre a la canalización.

Nota: A lo largo de este proceso, el esquema se propaga automáticamente a medida que compilas la canalización. También puedes importar un esquema deseado.

Eso es todo. Acabas de crear tu primera canalización de datos en tiempo real para que se implemente y ejecute.

Envía mensajes a través de Cloud Pub/Sub.

Con la IU de Pub/Sub:

1. Navega a la consola de GCP -> Pub/Sub -> Temas, selecciona tu tema y, luego, haz clic en PUBLICAR MENSAJE en el menú de la parte superior.

2. Coloca solo una fila de registro a la vez en el campo Mensaje. Haz clic en el botón +ADD AN ATTRIBUTE. Proporciona la clave = filename, valor = <type of record> (por ejemplo, pacientes, proveedores, alergias, etc.).
3. Haz clic en el botón Publicar para enviar el mensaje.

Usa el comando de gcloud:

1. Proporciona el mensaje de forma manual.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"paste one record row here"

2. Proporciona el mensaje de forma semiautomática con los comandos cat y sed de Unix. Este comando se puede ejecutar de manera repetida con diferentes parámetros.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"

Nota: <your-topic> - Igual que el tema proporcionado en las propiedades del nodo fuente de Pub/Sub

&lt;key&gt; - nombre del archivo

cat: Lee el contenido del archivo.

sed: Suprime el resultado (-n) y, luego, imprime (p) la línea específica #(1-?) de manera explícita. Si la fila de encabezado ya existe, usa "2p". etcétera.

6. Configura, implementa y ejecuta la canalización

Ahora que desarrollamos la canalización de datos, podemos implementarla y ejecutarla en Cloud Data Fusion.

1. Mantén los valores predeterminados en Configure.
2. Haz clic en Preview (Vista previa) para obtener una vista previa de los datos**.** Vuelve a hacer clic en **Preview** para volver a la ventana anterior. También puedes ejecutar la canalización en modo de vista previa haciendo clic en **EJECUTAR**.

3. Haz clic en Registros para ver los registros.
4. Haz clic en Guardar para guardar todos los cambios.
5. Haz clic en Import para importar la configuración de canalización guardada cuando compiles una canalización nueva.
6. Haz clic en Exportar para exportar una configuración de canalización.
7. Haz clic en Implementar para implementar la canalización.
8. Una vez implementada, haga clic en Ejecutar y espere que la canalización se ejecute hasta el final.

Nota: Cuando ejecutas una canalización, Cloud Data Fusion aprovisiona un clúster efímero de Cloud Dataproc, ejecuta la canalización y, luego, elimina el clúster. Esto puede tardar algunos minutos. Puedes observar el estado de la transición de la canalización de Aprovisionando a Iniciada y a En ejecución y a Desaprovisionando a Completada durante este tiempo.

9. Haz clic en Detener para detener la ejecución de la canalización en cualquier momento.
10. Si deseas duplicar la canalización, selecciona Duplicar en el botón Acciones.
11. Para exportar la configuración de la canalización, selecciona Exportar en el botón Acciones.

12. Haz clic en Resumen para ver gráficos del historial de ejecuciones, los registros, los registros de errores y las advertencias.

7. Validación

En esta sección, validaremos la ejecución de la canalización de datos.

1. Validar que la canalización se haya ejecutado correctamente y de forma continua

2. Validar que las tablas de BigQuery se carguen con registros actualizados basados en TIMESTAMP En este ejemplo, se publicaron dos mensajes o registros de pacientes y un mensaje o registro de alergias en el tema de Pub/Sub el 25-06-2019.

bq query --nouse_legacy_sql 'select (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Patients'  where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC" ) as Patients, (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Allergies' where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC") as Allergies;'

Waiting on bqjob_r14c8b94c1c0fe06a_0000016b960df4e1_1 ... (0s) Current status: DONE  
+----------+-----------+
| Patients | Allergies |
+----------+-----------+
|        2 |         1 |
+----------+-----------+

3. Valida que los mensajes publicados en <your-topic> fueron recibidos por <your-sub> suscriptor.

gcloud pubsub subscriptions pull --auto-ack <your-sub>

Cómo ver los resultados

Para ver los resultados después de publicar los mensajes en el tema de Pub/Sub mientras se ejecuta la canalización Realtime, haz lo siguiente:

1. Consultar la tabla en la IU de BigQuery IR A LA IU DE BIGQUERY
2. Actualiza la siguiente consulta con el nombre, el conjunto de datos y la tabla de tu proyecto.

8. Realice una limpieza

Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud Platform para los recursos que se usaron en este instructivo:

Una vez que hayas terminado el instructivo, puedes limpiar los recursos que creaste en GCP para que no consuman tu cuota y no se te facture por ellos en el futuro. En las secciones siguientes, se describe cómo borrar o desactivar estos recursos.

Borra el conjunto de datos de BigQuery

Sigue estas instrucciones para borrar el conjunto de datos de BigQuery que creaste como parte de este instructivo.

Borra el bucket de GCS

Sigue estas instrucciones para borrar el bucket de GCS que creaste como parte de este instructivo.

Borra la instancia de Cloud Data Fusion

Sigue estas instrucciones para borrar tu instancia de Cloud Data Fusion.

Borra el proyecto

La manera más fácil de eliminar la facturación es borrar el proyecto que creaste para el instructivo.

Para borrar el proyecto, haz lo siguiente:

Precaución: Borrar un proyecto tiene las siguientes consecuencias:

• Se borra todo en el proyecto. Si utilizaste un proyecto existente para este instructivo, cuando lo borres, también borrarás cualquier otro trabajo que hayas realizado en el proyecto.
• Se pierden los ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que utilizan el ID del proyecto, como una URL de appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.

Si planeas explorar varios instructivos y guías de inicio rápido, la reutilización de proyectos puede ayudarte a evitar exceder los límites de las cuotas del proyecto.

1. En GCP Console, ve a la página Proyectos. IR A LA PÁGINA PROYECTOS
2. En la lista de proyectos, selecciona el proyecto que deseas borrar y haz clic en Borrar.
3. En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.

9. Felicitaciones

Felicitaciones, completaste correctamente el codelab para transferir datos de atención médica a BigQuery con Cloud Data Fusion.

Publicaste datos de CSV en el tema de Pub/Sub y, luego, los cargaste en BigQuery.

Compilaste visualmente una canalización de integración de datos para cargar, transformar y enmascarar datos de atención médica en tiempo real.

Ahora conoces los pasos clave necesarios para comenzar tu recorrido de análisis de datos de atención médica con BigQuery en Google Cloud.