Transmite registros de Avro a BigQuery con Dataflow

En este instructivo, se describe cómo almacenar objetos SpecificRecord de Avro en BigQuery mediante Dataflow por medio de la generación automática del esquema de tabla y la transformación de los elementos de entrada. También se muestra el uso de las clases generadas por Avro para materializar o transmitir datos intermedios entre los trabajadores de tu canalización de Dataflow.

Apache Avro es un sistema de serialización que se basa en esquemas para estructurar datos. Debido a que el esquema siempre está presente cuando se leen o escriben datos de Avro, la serialización es rápida y breve. Los beneficios de rendimiento lo convierten en una opción popular para transmitir mensajes entre sistemas, como una app que envía eventos a un sistema de estadísticas por medio de un agente de mensajes. Puedes usar el esquema de Avro para administrar tu esquema de almacén de datos de BigQuery. Para convertir el esquema de Avro en la estructura de tabla de BigQuery, se requiere un código personalizado, como se muestra en este instructivo.

Este instructivo está dirigido a los desarrolladores y arquitectos interesados en usar el esquema de Avro para administrar el esquema de almacén de datos de BigQuery. En él, se supone que estás familiarizado con Avro y Java.

En el siguiente diagrama, se ilustra la arquitectura de alto nivel de este instructivo.

En este instructivo, se usa un sistema de procesamiento de pedidos simple con los siguientes pasos para demostrar este patrón arquitectónico:

• Una aplicación en línea genera eventos cuando el cliente realiza una compra.
• Un objeto de pedido contiene un identificador único, la lista de artículos comprados y una marca de tiempo.
• Una canalización de Dataflow lee los mensajes de Avro OrderDetails SpecificRecord desde un tema de Pub/Sub.
• La canalización de Dataflow escribe los registros en Cloud Storage como archivos de Avro.
• La clase OrderDetails genera de forma automática el esquema de BigQuery correspondiente.
• Los objetos OrderDetails se escriben en BigQuery mediante una función de transformación genérica.

Objetivos

• Transferir strings JSON desde una transmisión de datos de Pub/Sub mediante Dataflow
• Transformar los objetos JSON en objetos de clases generadas por Avro
• Generar el esquema de tabla de BigQuery a partir del esquema Avro
• Escribir los registros de Avro en un archivo de Cloud Storage
• Escribir los registros de Avro en BigQuery

Costos

En este instructivo, se usan los siguientes componentes facturables de Google Cloud:

• Pub/Sub
• Dataflow
• BigQuery
• Cloud Storage

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios. Es posible que los usuarios nuevos de Google Cloud califiquen para obtener una prueba gratuita.

Cuando finalices este instructivo, podrás borrar los recursos creados para evitar que se te siga facturando. Para obtener más información, consulta Cómo realizar una limpieza.

Antes de comenzar

1. Accede a tu cuenta de Google Cloud. Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

2. En la página del selector de proyectos de Google Cloud Console, selecciona o crea un proyecto de Google Cloud.

   Nota: Si no planeas conservar los recursos que creaste durante este procedimiento, crea un proyecto en lugar de seleccionar uno existente. Cuando termines, puedes borrar el proyecto y quitar todos los recursos asociados con él.

   Ir al selector de proyectos

3. Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.

4. Habilita las API de BigQuery, Cloud Storage, and Dataflow.

   Habilita las API

5. En la página del selector de proyectos de Google Cloud Console, selecciona o crea un proyecto de Google Cloud.

   Nota: Si no planeas conservar los recursos que creaste durante este procedimiento, crea un proyecto en lugar de seleccionar uno existente. Cuando termines, puedes borrar el proyecto y quitar todos los recursos asociados con él.

   Ir al selector de proyectos

6. Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.

7. Habilita las API de BigQuery, Cloud Storage, and Dataflow.

   Habilita las API

8. En la consola, activa Cloud Shell.

   Activar Cloud Shell

   En la parte inferior de la consola de Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

Configura tu entorno

1. En Cloud Shell, clona el repositorio de código fuente:

   cd $HOME && git clone https://github.com/GoogleCloudPlatform/bigquery-ingest-avro-dataflow-sample.git
   

2. Abre el archivo env.sh:

   cd $HOME/bigquery-ingest-avro-dataflow-sample
   nano env.sh
   

3. El archivo env.sh contiene ajustes predeterminados que puedes usar en este instructivo, pero puedes modificarlos para tu entorno.

   # pubsub topic
   MY_TOPIC="avro-records"
   
   # Cloud Storage Bucket
   MY_BUCKET="${GOOGLE_CLOUD_PROJECT}_avro_beam"
   
   # Avro file Cloud Storage output path
   AVRO_OUT="${MY_BUCKET}/out/"
   
   # Region for Cloud Pub/Sub and Cloud Dataflow
   REGION="us-central1"
   
   # Region for BigQuery
   BQ_REGION="US"
   
   # BigQuery dataset name
   BQ_DATASET="sales"
   
   # BigQuery table name
   BQ_TABLE="orders"
   

   Reemplaza los siguientes valores:

   • avro-records: Es el nombre de tu tema de Pub/Sub.
   • $GOOGLE_CLOUD_PROJECT"_avro_beam: Es el nombre de tu depósito de Cloud Storage que genera el ID del proyecto de Cloud.
   • $MY_BUCKET/""out/": Es la ruta del depósito de Cloud Storage que contiene tu resultado de Avro.
   • us-central1: Es la región que usas para Pub/Sub y Dataflow. Para obtener más información acerca de las regiones, consulta Geografía y regiones.
   • US: Es la región para BigQuery. Para obtener más información acerca de las ubicaciones, consulta Ubicaciones de conjuntos de datos.
   • sales: Es el nombre del conjunto de datos de BigQuery.
   • orders: el nombre de la tabla de BigQuery.
   • 1: la cantidad máxima de trabajadores de Dataflow.

4. Configura las variables de entorno:

    . ./env.sh
   

Crea recursos

1. En Cloud Shell, crea un tema de Pub/Sub:

   gcloud pubsub topics create "${MY_TOPIC}"
   

2. Crea un bucket de Cloud Storage:

   gsutil mb -l "${REGION}" -c regional "gs://${MY_BUCKET}"
   

   El depósito de Cloud Storage crea una copia de seguridad de los eventos sin procesar generados por la app. También puede servir como una fuente alternativa para la validación y las estadísticas sin conexión mediante el uso de trabajos de Spark y Hadoop ejecutados en Dataproc.

3. Crea un conjunto de datos de BigQuery:

   bq --location="${BQ_REGION}" mk --dataset "${GOOGLE_CLOUD_PROJECT}:${BQ_DATASET}"
   

   Un conjunto de datos de BigQuery contiene tablas y vistas en una sola región o una geografía que contiene varias regiones. Para obtener más información, consulta Crea conjuntos de datos.

Inicia la app de Beam Dataflow

1. En Cloud Shell, implementa y ejecuta la canalización en el ejecutor de Dataflow:

   cd $HOME/bigquery-ingest-avro-dataflow-sample/BeamAvro
   
   mvn clean package
   
   java -cp target/BeamAvro-bundled-1.0-SNAPSHOT.jar \
   com.google.cloud.solutions.beamavro.AvroToBigQuery \
   --runner=DataflowRunner \
   --project="${GOOGLE_CLOUD_PROJECT}" \
   --stagingLocation="gs://${MY_BUCKET}/stage/" \
   --tempLocation="gs://${MY_BUCKET}/temp/" \
   --inputPath="projects/${GOOGLE_CLOUD_PROJECT}/topics/${MY_TOPIC}" \
   --workerMachineType=n1-standard-1 \
   --region="${REGION}" \
   --dataset="${BQ_DATASET}" \
   --bigQueryTable="${BQ_TABLE}" \
   --outputPath="gs://${MY_BUCKET}/out/" \
   --jsonFormat=false \
   --avroSchema="$(<../orderdetails.avsc)"
   

   El resultado contiene el ID de tu app. Anótalo porque lo necesitarás más adelante en el instructivo.

2. En la consola, ve a Dataflow.

   Ir a Dataflow

3. Para ver el estado de la canalización, haz clic en el ID de tu app. El estado de la canalización se muestra como un gráfico.

   

Revisa el código

En el archivo AvroToBigQuery.java, las opciones de canalización con los parámetros requeridos pasan por los parámetros de la línea de comandos. La opción de modo de transmisión también está habilitada. El esquema de tabla de BigQuery se genera de forma automática a partir del esquema de Avro mediante Beam Schema de BigQuery IO

En el caso del formato de entrada de Avro, los objetos se leen desde Pub/Sub. Si el formato de entrada es JSON, los eventos se leen y se transforman en objetos de Avro.

BeamAvro/src/main/java/com/google/cloud/solutions/beamavro/AvroToBigQuery.java

Ver en GitHub

Schema avroSchema = new Schema.Parser().parse(options.getAvroSchema());

if (options.getJsonFormat()) {
  return input
      .apply("Read Json", PubsubIO.readStrings().fromTopic(options.getInputPath()))
      .apply("Make GenericRecord", MapElements.via(JsonToAvroFn.of(avroSchema)));
} else {
  return input.apply("Read GenericRecord", PubsubIO.readAvroGenericRecords(avroSchema)
      .fromTopic(options.getInputPath()));
}

La canalización se bifurca para realizar dos escrituras separadas:

• La transformación Write AVRO file escribe registros AVRO en Cloud Storage mediante AvroIO mediante la fragmentación de los datos en ventanas de 10 segundos:

  BeamAvro/src/main/java/com/google/cloud/solutions/beamavro/AvroToBigQuery.java

  Ver en GitHub

  // Write to GCS
  records
      .apply("Window for 10 seconds", Window.into(FixedWindows.of(Duration.standardSeconds(10))))
      .apply(
          "Write Avro file",
          AvroIO.writeGenericRecords(avroSchema).to(options.getOutputPath())
              .withWindowedWrites().withNumShards(5));

• La transformación Write to BigQuery escribe los registros en la tabla de BigQuery con Beam Schemas:

  BeamAvro/src/main/java/com/google/cloud/solutions/beamavro/AvroToBigQuery.java

  Ver en GitHub

  // Write to BigQuery
  records.apply(
      "Write to BigQuery",
      BigQueryIO.<GenericRecord>write()
          .to(bigQueryTable)
          .useBeamSchema()
          .withWriteDisposition(WriteDisposition.WRITE_APPEND)
          .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
          .optimizedWrites());

El BigQueryIO escribe los objetos de Avro en BigQuery y los transforma de forma interna en objetos TableRow usando esquemas de Beam. Consulta el asignación entre tipos de datos de BigQuery y tipos de datos de Avro.

Ver resultados en BigQuery

Para probar la canalización, inicia la secuencia de comandos gen.py. Esta secuencia de comandos simula la generación de eventos de pedido y los envía al tema de Pub/Sub.

1. En Cloud Shell, cambia al directorio de secuencia de comandos del generador de eventos de muestra y ejecuta la secuencia de comandos:

   cd $HOME/bigquery-ingest-avro-dataflow-sample/generator
   python3 -m venv env
   . ./env/bin/activate
   pip install -r requirements.txt
   python3 gen.py -p $GOOGLE_CLOUD_PROJECT -t $MY_TOPIC -n 100 -f avro
   

2. En la consola, ve a BigQuery.

   Ir a BigQuery

3. Para ver el esquema de tabla, haz clic en el conjunto de datos sales y, luego, selecciona la tabla orders. Si modificaste las variables de entorno predeterminadas en env.sh, los nombres del conjunto de datos y de la tabla pueden ser diferentes.

   

4. Para ver algunos datos de muestra, ejecuta una consulta en el Editor de consultas:

   SELECT * FROM sales.orders LIMIT 5
   

   

   El esquema de tabla de BigQuery se genera de forma automática a partir de los registros de Avro, y los datos se convierten de forma automática en la estructura de tabla de BigQuery.

Limpia

Borra el proyecto

Precaución: Borrar un proyecto tiene estas 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 appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.
1. En la consola, ve a la página Administrar recursos.

   Ir a Administrar recursos

2. En la lista de proyectos, elige el proyecto que quieres borrar y haz clic en Borrar.
3. En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrar el proyecto.

Borra los recursos individuales

1. Sigue estas instrucciones para detener el trabajo de Dataflow.

2. Borra el bucket de Cloud Storage:

   gsutil rm -r gs://$MY_BUCKET
   

¿Qué sigue?

• Obtén información sobre los patrones comunes de casos prácticos de Dataflow.
• Descubre cómo puedes usar Dataflow con un clúster de Apache Kafka alojado fuera de Google Cloud.
• Explora arquitecturas de referencia, diagramas, instructivos y prácticas recomendadas sobre Google Cloud. Consulta nuestro Cloud Architecture Center.