Transfiere datos CSV a BigQuery con Cloud Data Fusion: transferencia por lotes

1. Introducción

Última actualización: 28/02/2020

En este codelab, se demuestra un patrón de transferencia de datos para transferir de forma masiva datos de atención médica con formato CSV a BigQuery. En este lab, usaremos la canalización de datos por lotes de Cloud Data Fusion. Se generaron datos de prueba de atención médica realistas y se pusieron a tu 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 CSV (carga programada por lotes) de GCS a BigQuery con Cloud Data Fusion
• Cómo crear visualmente una canalización de integración de datos en Cloud Data Fusion para cargar, transformar y enmascarar datos de atención médica de forma masiva

¿Qué necesitas para ejecutar este codelab?

• Necesitas acceso a un proyecto de GCP.
• Se te debe asignar el rol de propietario del proyecto de GCP.
• Son 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 CSV de recursos tiene su propia 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úbelo a GCS en lugar de copiarlo del bucket en el paso Copia los datos de entrada.

Referencia:

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

2. Configuración del proyecto de GCP

Inicializa las variables de shell para tu entorno.

Para encontrar el PROJECT_ID, consulta Cómo identificar 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 los datos de entrada y los registros de errores con la herramienta gsutil.

gsutil mb -l us gs://$BUCKET_NAME

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

1. Desde la dirección de correo electrónico que usas para acceder a Cloud Console, 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 para responder el correo electrónico y 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

Crea un conjunto de datos de BigQuery.

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

Recuerda: Siempre puedes generar un nuevo conjunto de datos de prueba realistas con SyntheaTM para este lab. Para tu comodidad, proporcionamos un conjunto de datos en GCS: gs://hcls_testing_data_fhir_10_patients/csv

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 APIs.

1. Ve a la Biblioteca de API de GCP Console.
2. Selecciona el proyecto desde la lista de proyectos.
3. En la biblioteca de API, selecciona la API que quieres habilitar. Si necesitas ayuda para encontrar la API, usa el campo de búsqueda o los filtros.
4. En la página de la API, haz clic en HABILITAR.

Crea una instancia de Cloud Data Fusion.

1. En GCP Console, selecciona tu ProjectID.
2. Selecciona Data Fusion en el menú de la izquierda y, luego, haz clic en el botón CREAR UNA INSTANCIA en el centro de la página (primera creación) o en el botón CREAR INSTANCIA 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 instancias puede tardar hasta 30 minutos en completarse.

Configura los permisos de la instancia.

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

Nota: El permiso solo debe otorgarse 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 de IAM de tu proyecto.
4. En la página Permisos de IAM, ahora agregaremos la cuenta de servicio como un miembro nuevo y le otorgaremos el rol Agente de servicio de la API de Cloud Data Fusion. Haz clic en el botón Agregar, pega la "cuenta de servicio" en el campo Miembros nuevos y selecciona el rol Agente de servicio de la API de Cloud Data Fusion -> Administración de servicio.
5. 
6. Haz clic en Guardar.

Una vez que completes estos pasos, puedes comenzar a usar Cloud Data Fusion. Para ello, haz 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 GCP Console -> Red de VPC -> Reglas de firewall para verificar si existe la regla default-allow-ssh.

2. Si no es así, agrega una regla de firewall que permita todo el tráfico SSH de entrada a la red predeterminada.

Usa 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

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

4. Cómo compilar un esquema para la transformación

Ahora que tenemos el entorno de Cloud Fusion en GCP, vamos a compilar un esquema. Necesitamos este esquema para transformar los datos CSV.

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. Tu elección de hacer clic en el botón "Iniciar recorrido" o "No, gracias" en la ventana emergente de bienvenida
2. Expande el menú "hamburguesa" y selecciona Pipeline -> Studio.

3. En la sección Transform de la paleta de complementos que se encuentra a la izquierda, haz doble clic en el nodo Wrangler, que aparecerá en la IU de Data Pipelines.

4. Selecciona el nodo Wrangler y haz clic en Propiedades. Haz clic en el botón Wrangle y, luego, selecciona un archivo fuente .csv (por ejemplo, patients.csv), que debe tener todos los campos de datos para compilar el esquema deseado.
5. Haz clic en la flecha hacia abajo (Transformaciones de columnas) junto a cada nombre de columna (por ejemplo, body).
6. De forma predeterminada, la importación inicial supondrá que solo hay una columna en tu archivo de datos. Para analizarlo como un archivo CSV, elige Parse → CSV y, luego, selecciona el delimitador y marca la casilla "Set first row as header" según corresponda. Haz clic en el botón Aplicar.
7. Haz clic en la flecha hacia abajo junto al campo Cuerpo y selecciona Borrar columna para quitar el campo Cuerpo. Además, puedes probar otras transformaciones, como quitar columnas, cambiar el tipo de datos de algunas columnas (el valor predeterminado es el tipo "cadena"), dividir columnas, establecer nombres de columnas, etcétera.

8. Las pestañas "Columnas" y "Pasos de transformación" muestran el esquema de salida y la receta de Wrangler. Haz clic en Aplicar en la esquina superior derecha. Haz clic en el botón Validar. El mensaje verde "No se encontraron errores" indica que la operación se realizó correctamente.

9. En Wrangler Properties, haz clic en el menú desplegable Actions para Export el esquema deseado en tu almacenamiento local para una futura Import si es necesario.
10. Guarda la receta de Wrangler para usarla en el futuro.

parse-as-csv :body ',' true
drop body

11. Para cerrar la ventana Wrangler Properties, haz clic en el botón X.

5. Compila nodos para la canalización

En esta sección, compilaremos los componentes de la canalización.

1. En la esquina superior izquierda de la IU de Data Pipelines, deberías ver que Data Pipeline - Batch está seleccionado como el tipo de canalización.

2. En el panel izquierdo, hay diferentes secciones, como Filter, Source, Transform, Analytics, Sink, Conditions and Actions, Error Handlers y Alerts, en las que puedes seleccionar uno o varios nodos para la canalización.

Nodo fuente

1. Selecciona el nodo Source.
2. En la sección Source de la paleta Plugin que se encuentra a la izquierda, haz doble clic en el nodo Google Cloud Storage, que aparece en la IU de Data Pipelines.
3. Selecciona el nodo fuente de GCS y haz clic en Propiedades.

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

• Label = {cualquier texto}
• Nombre de referencia = {cualquier texto}
• ID del proyecto = detección automática
• Ruta de acceso = URL de GCS al bucket en tu proyecto actual. Por ejemplo, gs://$BUCKET_NAME/csv/
• Formato = texto
• Path Field = filename
• Path Filename Only = true
• Read Files Recursively = true

5. Haz clic en el botón + para agregar el campo "filename" al esquema de salida de GCS.
6. Haz clic en Documentación para obtener una explicación detallada. Haz clic en el botón Validar. El mensaje verde "No se encontraron errores" indica que la operación se realizó correctamente.
7. Para cerrar las propiedades de GCS, haz clic en el botón X.

Nodo de transformación

1. Selecciona el nodo Transform.
2. En la sección Transform de la paleta de complementos que se encuentra a la izquierda, haz doble clic en el nodo Wrangler, que aparece en la IU de Data Pipelines. Conecta el nodo fuente de GCS al nodo de transformación de Wrangler.
3. Selecciona el nodo Wrangler y haz clic en Propiedades.
4. 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/ schema (Patients).json) y pega la receta guardada de la sección anterior.
5. También puedes reutilizar el nodo Wrangler de la sección Crea un esquema para la transformación.
6. Llena los campos obligatorios. Configura los siguientes campos:

• Label = {cualquier texto}
• Nombre del campo de entrada = {*}
• Precondition = {filename != "patients.csv"} para distinguir cada archivo de entrada (por ejemplo, patients.csv, providers.csv, allergies.csv, etc.) del nodo Source.

7. Agrega un nodo de JavaScript para ejecutar el código JavaScript proporcionado por el usuario que transforma aún más los registros. En este codelab, utilizamos el nodo JavaScript para obtener una marca de tiempo para cada actualización de registro. Conecta el nodo de transformación de Wrangler al nodo de transformación de JavaScript. Abre Propiedades de JavaScript y agrega la siguiente función:

function transform(input, emitter, context) {
  input.TIMESTAMP = (new Date()).getTime()*1000;
  emitter.emit(input);
}

8. Haz clic en el signo + para agregar el campo llamado TIMESTAMP al esquema de salida (si no existe). Selecciona la marca de tiempo como el tipo de datos.

9. Haz clic en Documentación para obtener una explicación detallada. Haz clic en el botón Validar para validar toda la información de entrada. El mensaje verde "No se encontraron errores" indica que la operación se realizó correctamente.
10. Para cerrar la ventana Transform Properties, haz clic en el botón X.

Enmascaramiento y desidentificación de datos

1. Puedes seleccionar columnas de datos individuales haciendo clic en la flecha hacia abajo de la columna y aplicando reglas de enmascaramiento en la selección Enmascarar datos según tus requisitos (por ejemplo, la columna de número de seguridad social).

2. Puedes agregar más directivas en la ventana Recipe del nodo Wrangler. Por ejemplo, usar la directiva hash con el algoritmo de hash siguiendo esta sintaxis para fines de desidentificación:

hash <column> <algorithm> <encode>

<column>: name of the column
<algorithm>: Hashing algorithm (i.e. MD5, SHA-1, etc.)
<encode>: default is true (hashed digest is encoded as hex with left-padding zeros). To disable hex encoding, set <encode> to false.

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

Nodo receptor

1. Selecciona el nodo receptor.
2. En la sección Sink de la paleta de complementos que se encuentra a la izquierda, haz doble clic en el nodo BigQuery, que aparecerá en la IU de Data Pipeline.
3. Selecciona el nodo receptor de BigQuery y haz clic en Propiedades.

4. Completa los campos obligatorios. Configura los siguientes campos:

• Label = {cualquier texto}
• Nombre de referencia = {cualquier texto}
• ID del proyecto = detección automática
• Conjunto de datos = Conjunto de datos de BigQuery que se usa en el proyecto actual (es decir, DATASET_ID)
• Table = {nombre de la tabla}

5. Haz clic en Documentación para obtener una explicación detallada. Haz clic en el botón Validar para validar toda la información de entrada. El mensaje verde "No se encontraron errores" indica que la operación se realizó correctamente.

6. Para cerrar las propiedades de BigQuery, haz clic en el botón X.

6. Compila una canalización de datos por lotes

Cómo conectar todos los nodos de una canalización

1. Arrastra una flecha de conexión > del borde derecho del nodo fuente y suéltala en el borde izquierdo del nodo de destino.
2. Una canalización puede tener varias ramas que obtienen archivos de entrada del mismo nodo de origen de GCS.

Nota: Compila una rama desde el nodo fuente en la canalización para cada archivo .csv de entrada.

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 el esquema deseado.

Eso es todo. Acabas de crear tu primera canalización de datos por lotes y puedes implementarla y ejecutarla.

Envía alertas de la canalización por correo electrónico (opcional)

Para utilizar la función SendEmail de Pipeline Alert, la configuración requiere que se configure un servidor de correo para enviar correos desde una instancia de máquina virtual. Consulta el siguiente vínculo de referencia para obtener más información:

Envía correos electrónicos desde una instancia | Documentación de Compute Engine

En este codelab, configuraremos un servicio de retransmisión de correo a través de Mailgun con los siguientes pasos:

1. Sigue las instrucciones en Envía correos electrónicos con Mailgun | Documentación de Compute Engine para configurar una cuenta con Mailgun y configurar el servicio de retransmisión de correo electrónico. A continuación, se indican las modificaciones adicionales.
2. Agrega todas las direcciones de correo electrónico de los destinatarios a la lista de autorizados de Mailgun. Puedes encontrar esta lista en la opción Mailgun> Sending> Overview del panel izquierdo.

Una vez que los destinatarios hagan clic en "Acepto" en el correo electrónico enviado desde [e-mail], sus direcciones de correo electrónico se guardarán en la lista de direcciones autorizadas para recibir correos electrónicos de alerta de la canalización.

3. Paso 3 de la sección "Antes de comenzar": Crea una regla de firewall de la siguiente manera:

4. Paso 3 de "Configura Mailgun como retransmisión de correo con Postfix". Selecciona Sitio de Internet o Internet con host inteligente, en lugar de Solo local, como se menciona en las instrucciones.

5. Paso 4 de "Configura Mailgun como retransmisión de correo con Postfix". Edita vi /etc/postfix/main.cf para agregar 10.128.0.0/9 al final de mynetworks.

6. Edita vi /etc/postfix/master.cf para cambiar el puerto smtp predeterminado (25) al puerto 587.

7. En la esquina superior derecha de Data Fusion Studio, haz clic en Configurar. Haz clic en Alerta de canalización y, luego, en el botón + para abrir la ventana Alertas. Selecciona SendEmail.

8. Completa el formulario de configuración de Correo electrónico. Selecciona completion, success o failure en el menú desplegable Run Condition para cada tipo de alerta. Si Include Workflow Token = false, solo se envía la información del campo Message. Si Include Workflow Token = true, se envía la información del campo Message y la información detallada del token de flujo de trabajo. Debes usar minúsculas para Protocol. Usa cualquier correo electrónico "falso" que no sea la dirección de correo electrónico de tu empresa para el Remitente.

7. Configura, implementa, ejecuta o programa la canalización

1. En la esquina superior derecha de Data Fusion Studio, haz clic en Configurar. Selecciona Spark en Engine Config. Haz clic en Guardar en la ventana Configurar.

2. Haz clic en Vista previa para obtener una vista previa de los datos y, luego, vuelve a hacer clic en **Vista previa** para volver a la ventana anterior. También puedes **ejecutar** la canalización en el modo de vista previa.

3. Haz clic en Registros para ver los registros.
4. Haz clic en Guardar para guardar todos los cambios.
5. Haz clic en Importar para importar la configuración de la canalización guardada cuando compiles una canalización nueva.
6. Haz clic en Exportar para exportar la configuración de una 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. Durante este tiempo, puedes observar el estado de la transición de la canalización de Aprovisionando a Iniciada a En ejecución a Desaprovisionando a Completada.

9. Para duplicar la canalización, selecciona Duplicar en el botón Acciones.
10. Puedes exportar la configuración de la canalización seleccionando Exportar en el botón Acciones.
11. Haz clic en Activadores de entrada o Activadores de salida en el borde izquierdo o derecho de la ventana de Studio para establecer activadores de canalización si lo deseas.
12. Haz clic en Programar para programar la ejecución de la canalización y cargar datos de forma periódica.

13. En Resumen, se muestran gráficos del historial de ejecuciones, los registros, los registros de errores y las advertencias.

8. Validación

1. La canalización de Validate se ejecutó correctamente.

2. Valida si el conjunto de datos de BigQuery tiene todas las tablas.

bq ls $PROJECT_ID:$DATASET_ID

     tableId       Type    Labels   Time Partitioning
----------------- ------- -------- -------------------
 Allergies         TABLE
 Careplans         TABLE
 Conditions        TABLE
 Encounters        TABLE
 Imaging_Studies   TABLE
 Immunizations     TABLE
 Medications       TABLE
 Observations      TABLE
 Organizations     TABLE
 Patients          TABLE
 Procedures        TABLE
 Providers         TABLE

3. Recibir correos electrónicos de alerta (si se configuraron)

Cómo ver los resultados

Para ver los resultados después de la ejecución de una canalización, haga lo siguiente:

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

9. Realiza 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 completes 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.

Cómo borrar 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 URLs que usan 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, elige 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.

10. Felicitaciones

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

Importaste datos CSV de Google Cloud Storage a BigQuery.

Creaste visualmente la canalización de integración de datos para cargar, transformar y enmascarar datos de atención médica de forma masiva.

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.