Migrar esquemas y datos desde Apache Hive
En este documento se describe cómo migrar sus datos, ajustes de seguridad y flujos de trabajo de Apache Hive a BigQuery.
También puedes usar la traducción de SQL por lotes para migrar tus secuencias de comandos SQL en bloque o la traducción de SQL interactiva para traducir consultas específicas. Los servicios de traducción de SQL admiten Apache HiveQL.
Preparar la migración
En las siguientes secciones se describe cómo recoger información sobre las estadísticas, los metadatos y la configuración de seguridad de las tablas para ayudarle a migrar su almacén de datos de Hive a BigQuery.
Recoger información de la tabla de origen
Recopila información sobre las tablas de Hive de origen, como el número de filas, el número de columnas, los tipos de datos de las columnas, el tamaño, el formato de entrada de los datos y la ubicación. Esta información es útil en el proceso de migración y también para validar la migración de datos. Si tienes una tabla de Hive llamada employees en una base de datos llamada corp, usa los siguientes comandos para recoger información sobre la tabla:
# Find the number of rows in the table hive> SELECT COUNT(*) FROM corp.employees; # Output all the columns and their data types hive> DESCRIBE corp.employees; # Output the input format and location of the table hive> SHOW CREATE TABLE corp.employees; Output: … STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat' LOCATION 'hdfs://demo_cluster/user/hive/warehouse/corp/employees' TBLPROPERTIES ( … # Get the total size of the table data in bytes shell> hdfs dfs -du -s TABLE_LOCATION
Conversión del formato de la tabla de origen
Algunos de los formatos que admite Hive no se pueden ingerir directamente en BigQuery.
Hive admite el almacenamiento de datos en los siguientes formatos:
- Archivo de texto
- Archivo RC
- Archivo de secuencia
- Archivo Avro
- Archivo ORC
- Archivo Parquet
BigQuery admite la carga de datos de Cloud Storage en cualquiera de los siguientes formatos de archivo:
- CSV
- JSON (delimitado por líneas nuevas)
- Avro
- ORC
- Parquet
BigQuery puede cargar archivos de datos en formatos Avro, ORC y Parquet directamente sin necesidad de archivos de esquema. En el caso de los archivos de texto que no tienen el formato CSV o JSON (delimitado por líneas nuevas), puedes copiar los datos en una tabla de Hive con formato Avro o convertir el esquema de la tabla en un esquema JSON de BigQuery para proporcionarlo al ingerir los datos.
Recoger los ajustes de control de acceso de Hive
Hive y BigQuery tienen mecanismos de control de acceso diferentes. Recopila todos los ajustes de control de acceso de Hive, como los roles, los grupos, los miembros y los privilegios que se les han concedido. Define un modelo de seguridad en BigQuery a nivel de conjunto de datos e implementa una lista de control de acceso granular. Por ejemplo, un usuario de Hive se puede asignar a una cuenta de Google y un grupo de HDFS se puede asignar a un grupo de Google. El acceso se puede definir a nivel del conjunto de datos. Usa los siguientes comandos para recoger la configuración de control de acceso en Hive:
# List all the users > hdfs dfs -ls /user/ | cut -d/ -f3 # Show all the groups that a specific user belongs to > hdfs groups user_name # List all the roles hive> SHOW ROLES; # Show all the roles assigned to a specific group hive> SHOW ROLE GRANT GROUP group_name # Show all the grants for a specific role hive> SHOW GRANT ROLE role_name; # Show all the grants for a specific role on a specific object hive> SHOW GRANT ROLE role_name on object_type object_name;
En Hive, puedes acceder directamente a los archivos HDFS que hay detrás de las tablas si tienes los permisos necesarios. En las tablas estándar de BigQuery, una vez que los datos se cargan en la tabla, se almacenan en el almacenamiento de BigQuery. Puede leer datos mediante la API Read de BigQuery Storage, pero se sigue aplicando la seguridad a nivel de gestión de identidades y accesos, de fila y de columna. Si usas tablas externas de BigQuery para consultar los datos de Cloud Storage, el acceso a Cloud Storage también se controla mediante IAM.
Puedes crear una tabla de BigLake que te permita usar conectores para consultar los datos con Apache Spark, Trino o Apache Hive. La API Storage de BigQuery aplica políticas de control a nivel de fila y de columna a todas las tablas BigLake de Cloud Storage o BigQuery.
Migración de datos
La migración de datos de Hive desde tu clúster de origen local u otro clúster de origen basado en la nube a BigQuery consta de dos pasos:
- Copiar datos de un clúster de origen en Cloud Storage
- Cargar datos de Cloud Storage en BigQuery
En las siguientes secciones se explica cómo migrar datos de Hive, validar los datos migrados y gestionar la migración de datos insertados continuamente. Los ejemplos se han escrito para tablas que no son ACID.
Datos de la columna de partición
En Hive, los datos de las tablas con particiones se almacenan en una estructura de directorios.
Cada partición de la tabla se asocia a un valor concreto de la columna de partición. Los archivos de datos no contienen ningún dato de las columnas de partición. Usa el comando SHOW PARTITIONS para enumerar las diferentes particiones de una tabla con particiones.
En el ejemplo siguiente se muestra que la tabla de origen de Hive tiene particiones en las columnas joining_date y department. Los archivos de datos de esta tabla no contienen datos relacionados con estas dos columnas.
hive> SHOW PARTITIONS corp.employees_partitioned joining_date="2018-10-01"/department="HR" joining_date="2018-10-01"/department="Analyst" joining_date="2018-11-01"/department="HR"
Una forma de copiar estas columnas es convertir la tabla particionada en una tabla sin particiones antes de cargarla en BigQuery:
- Crea una tabla sin particiones con un esquema similar al de la tabla con particiones.
- Carga los datos en la tabla sin particiones desde la tabla de origen con particiones.
- Copia estos archivos de datos en Cloud Storage desde la tabla no particionada provisional.
- Carga los datos en BigQuery con el comando
bq loady proporciona el nombre de la columna de partición de tipoTIMESTAMPoDATE, si la hay, como argumentotime_partitioning_field.
Copiar datos en Cloud Storage
El primer paso para migrar datos es copiarlos en Cloud Storage. Usa Hadoop DistCp para copiar datos de tu clúster local o de otra nube en Cloud Storage. Almacena los datos en un segmento de la misma región o multirregión que el conjunto de datos en el que quieras almacenarlos en BigQuery. Por ejemplo, si quieres usar un conjunto de datos de BigQuery que ya tengas como destino y que esté en la región de Tokio, debes elegir un segmento regional de Cloud Storage en Tokio para almacenar los datos.
Después de seleccionar la ubicación del segmento de Cloud Storage, puedes usar el siguiente comando para enumerar todos los archivos de datos presentes en la ubicación de la tabla de employees Hive:
> hdfs dfs -ls hdfs://demo_cluster/user/hive/warehouse/corp/employees hdfs://demo_cluster/user/hive/warehouse/corp/employees/000000_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000001_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000002_0
Copia todos los archivos anteriores en Cloud Storage:
> hadoop distcp hdfs://demo_cluster/user/hive/warehouse/corp/employees gs://hive_data/corp/employees
Ten en cuenta que se te cobrará por almacenar los datos en Cloud Storage según los precios del almacenamiento de datos.
Puede haber directorios de staging que contengan archivos intermedios creados para tareas de consulta. Debes asegurarte de eliminar todos los directorios de este tipo antes de ejecutar el comando bq load.
Cargando datos
BigQuery admite la carga por lotes de datos en muchos formatos desde Cloud Storage. Asegúrate de que el conjunto de datos de BigQuery en el que quieras cargar los datos exista antes de crear un trabajo de carga.
El siguiente comando muestra los datos copiados de Hive de una tabla que no es ACID:
> gcloud storage ls gs://hive_data/corp/employees/ gs://hive-migration/corp/employees/ gs://hive-migration/corp/employees/000000_0 gs://hive-migration/corp/employees/000001_0 gs://hive-migration/corp/employees/000002_0
Para cargar tus datos de Hive en BigQuery, usa el comando bq load.
Puedes usar el carácter comodín * en la URL para cargar datos de varios archivos que compartan un prefijo de objeto común. Por ejemplo, usa el siguiente comando para cargar todos los archivos que compartan el prefijo gs://hive_data/corp/employees/:
bq load --source_format=AVRO corp.employees gs://hive_data/corp/employees/*
Como los trabajos pueden tardar mucho en completarse, puedes ejecutarlos de forma asíncrona configurando la marca --sync en False. Al ejecutar el comando bq load
, se muestra el ID de la tarea de carga creada, por lo que puedes usar este comando
para sondear el estado de la tarea.
Estos datos incluyen detalles como el tipo de trabajo, el estado del trabajo y el usuario que lo ha ejecutado.
Consulta el estado de cada tarea de carga mediante su ID correspondiente y comprueba si alguna tarea ha fallado con errores. En caso de fallo, BigQuery usa un enfoque de "Todo o nada" al cargar datos en una tabla. Puedes intentar resolver los errores y volver a crear otro trabajo de carga de forma segura. Para obtener más información, consulta Solucionar errores.
Asegúrate de que tienes suficiente cuota de tareas de carga por tabla y proyecto. Si superas tu cuota, el trabajo de carga fallará y se devolverá un error quotaExceeded.
Ten en cuenta que no se te cobra por cargar datos en BigQuery desde Cloud Storage. Cuando los datos se cargan en BigQuery, pasan a estar sujetos a los precios de almacenamiento de ese servicio. Cuando los trabajos de carga se hayan completado correctamente, puedes eliminar los archivos que queden en Cloud Storage para evitar que se te cobren cargos por almacenar datos redundantes.
Validación
Una vez que hayas cargado los datos correctamente, puedes validar los datos migrados comparando el número de filas de las tablas de Hive y BigQuery. Consulte la información de la tabla para obtener detalles sobre las tablas de BigQuery, como el número de filas, el número de columnas, los campos de partición o los campos de clustering. Para llevar a cabo una validación adicional, prueba la herramienta de validación de datos.
Ingestión continua
Si ingieres datos continuamente en una tabla de Hive, realiza una migración inicial y, a continuación, migra solo los cambios de datos incrementales a BigQuery. Es habitual crear secuencias de comandos que se ejecuten repetidamente para buscar y cargar datos nuevos. Hay muchas formas de hacerlo, y en las siguientes secciones se describe un enfoque posible.
Puedes hacer un seguimiento del progreso de la migración en una tabla de base de datos de Cloud SQL, que se denomina tabla de seguimiento en las siguientes secciones. Durante la primera ejecución de la migración, almacena el progreso en la tabla de seguimiento. En las ejecuciones posteriores de la migración, usa la información de la tabla de seguimiento para detectar si se han ingerido datos adicionales y si se pueden migrar a BigQuery.
Seleccione una columna de identificador de tipo INT64, TIMESTAMP o DATE para distinguir los datos incrementales. Esto se conoce como columna incremental.
La siguiente tabla es un ejemplo de una tabla sin particiones que usa el tipo TIMESTAMP para su columna incremental:
+-----------------------------+-----------+-----------+-----------+-----------+ | timestamp_identifier | column_2 | column_3 | column_4 | column_5 | +-----------------------------+-----------+-----------+-----------+-----------+ | 2018-10-10 21\:56\:41 | | | | | | 2018-10-11 03\:13\:25 | | | | | | 2018-10-11 08\:25\:32 | | | | | | 2018-10-12 05\:02\:16 | | | | | | 2018-10-12 15\:21\:45 | | | | | +-----------------------------+-----------+-----------+-----------+-----------+
En la siguiente tabla se muestra un ejemplo de una tabla con particiones en una columna de tipo DATEpartition_column. Tiene una columna incremental de tipo entero int_identifier
en cada partición.
+---------------------+---------------------+----------+----------+-----------+ | partition_column | int_identifier | column_3 | column_4 | column_5 | +---------------------+---------------------+----------+----------+-----------+ | 2018-10-01 | 1 | | | | | 2018-10-01 | 2 | | | | | ... | ... | | | | | 2018-10-01 | 1000 | | | | | 2018-11-01 | 1 | | | | | 2018-11-01 | 2 | | | | | ... | ... | | | | | 2018-11-01 | 2000 | | | | +---------------------+---------------------+----------+----------+-----------+
En las siguientes secciones se describe cómo migrar datos de Hive en función de si están particionados y de si tienen columnas incrementales.
Tabla sin particiones y sin columnas incrementales
Si no hay compactaciones de archivos en Hive, Hive crea archivos de datos nuevos al ingerir datos nuevos. Durante la primera ejecución, almacena la lista de archivos en la tabla de seguimiento y completa la migración inicial de la tabla de Hive copiando estos archivos en Cloud Storage y cargándolos en BigQuery.
> hdfs dfs -ls hdfs://demo_cluster/user/hive/warehouse/corp/employees Found 3 items hdfs://demo_cluster/user/hive/warehouse/corp/employees/000000_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000001_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000002_0
Después de la migración inicial, algunos datos se ingieren en Hive. Solo tienes que migrar estos datos incrementales a BigQuery. En las migraciones posteriores, vuelva a enumerar los archivos de datos y compárelos con la información de la tabla de seguimiento para detectar los archivos de datos nuevos que no se hayan migrado.
> hdfs dfs -ls hdfs://demo_cluster/user/hive/warehouse/corp/employees Found 5 items hdfs://demo_cluster/user/hive/warehouse/corp/employees/000000_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000001_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000002_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000003_0 hdfs://demo_cluster/user/hive/warehouse/corp/employees/000004_0
En este ejemplo, hay dos archivos nuevos en la ubicación de la tabla. Migra los datos copiando estos nuevos archivos de datos en Cloud Storage y cargándolos en la tabla de BigQuery.
Tabla sin particiones con columnas incrementales
En este caso, puede usar el valor máximo de las columnas incrementales para determinar si se ha añadido algún dato nuevo. Mientras se realiza la migración inicial, consulta la tabla de Hive para obtener el valor máximo de la columna incremental y almacenarlo en la tabla de seguimiento:
hive> SELECT MAX(timestamp_identifier) FROM corp.employees; 2018-12-31 22:15:04
En las ejecuciones posteriores de la migración, repite la misma consulta para obtener el valor máximo actual de la columna incremental y compáralo con el valor máximo anterior de la tabla de seguimiento para comprobar si existen datos incrementales:
hive> SELECT MAX(timestamp_identifier) FROM corp.employees; 2019-01-04 07:21:16
Si el valor máximo actual es mayor que el anterior, significa que se han insertado datos incrementales en la tabla de Hive, como en el ejemplo. Para migrar los datos incrementales, crea una tabla de almacenamiento provisional y carga solo los datos incrementales en ella.
hive> CREATE TABLE stage_employees LIKE corp.employees; hive> INSERT INTO TABLE stage_employees SELECT * FROM corp.employees WHERE timestamp_identifier>"2018-12-31 22:15:04" and timestamp_identifier<="2019-01-04 07:21:16"
Migra la tabla de almacenamiento provisional enumerando los archivos de datos de HDFS, copiándolos en Cloud Storage y cargándolos en la tabla de BigQuery.
Tabla con particiones sin columnas incrementales
La ingestión de datos en una tabla con particiones puede crear particiones, añadir datos incrementales a particiones ya creadas o ambas cosas. En este caso, puedes identificar las particiones actualizadas, pero no puedes identificar fácilmente qué datos se han añadido a estas particiones, ya que no hay ninguna columna incremental que permita distinguirlos. Otra opción es crear y mantener copias de HDFS, pero la creación de copias genera problemas de rendimiento en Hive, por lo que suele estar inhabilitada.
Al migrar la tabla por primera vez, ejecute el comando SHOW PARTITIONS y almacene la información sobre las diferentes particiones en la tabla de seguimiento.
hive> SHOW PARTITIONS corp.employees partition_column=2018-10-01 partition_column=2018-11-01
El resultado anterior muestra que la tabla employees tiene dos particiones. A continuación, se muestra una versión simplificada de la tabla de seguimiento para mostrar cómo se puede almacenar esta información.
| partition_information | file_path | gcs_copy_status | gcs_file_path | bq_job_id | ... |
|---|---|---|---|---|---|
| partition_column =2018-10-01 | |||||
| partition_column =2018-11-01 |
En las migraciones posteriores, vuelve a ejecutar el comando SHOW PARTITIONS para enumerar todas las particiones y compararlas con la información de partición de la tabla de seguimiento para comprobar si hay alguna partición nueva que no se haya migrado.
hive> SHOW PARTITIONS corp.employees partition_column=2018-10-01 partition_column=2018-11-01 partition_column=2018-12-01 partition_column=2019-01-01
Si se identifican particiones nuevas como en el ejemplo, cree una tabla de almacenamiento temporal y cargue solo las particiones nuevas en ella desde la tabla de origen. Migra la tabla de almacenamiento provisional copiando los archivos en Cloud Storage y cargándolos en la tabla de BigQuery.
Tabla con particiones con columnas incrementales
En este caso, la tabla de Hive está particionada y hay una columna incremental en cada partición. Los datos insertados continuamente se incrementan en este valor de columna. Aquí puedes migrar las nuevas particiones como se describe en la sección anterior, así como los datos incrementales que se hayan insertado en las particiones existentes.
Cuando migre la tabla por primera vez, almacene los valores mínimo y máximo de la columna incremental de cada partición junto con la información sobre las particiones de la tabla en la tabla de seguimiento.
hive> SHOW PARTITIONS corp.employees partition_column=2018-10-01 partition_column=2018-11-01 hive> SELECT MIN(int_identifier),MAX(int_identifier) FROM corp.employees WHERE partition_column="2018-10-01"; 1 1000 hive> SELECT MIN(int_identifier),MAX(int_identifier) FROM corp.employees WHERE partition_column="2018-11-01"; 1 2000
El resultado anterior muestra que la tabla employees tiene dos particiones y los valores mínimo y máximo de la columna incremental de cada partición. A continuación, se muestra una versión simplificada de la tabla de seguimiento para mostrar cómo se puede almacenar esta información.
| partition_information | inc_col_min | inc_col_max | file_path | gcs_copy_status | ... |
|---|---|---|---|---|---|
| partition_column =2018-10-01 | 1 | 1000 | |||
| partition_column =2018-11-01 | 1 | 2000 |
En las ejecuciones posteriores, ejecuta las mismas consultas para obtener el valor máximo actual de cada partición y compáralo con el valor máximo anterior de la tabla de seguimiento.
hive> SHOW PARTITIONS corp.employees partition_column=2018-10-01 partition_column=2018-11-01 partition_column=2018-12-01 partition_column=2019-01-01 hive> SELECT MIN(int_identifier),MAX(int_identifier) FROM corp.employees WHERE partition_column="2018-10-01";
En el ejemplo, se han identificado dos particiones nuevas y se han ingerido algunos datos incrementales en la partición partition_column=2018-10-01.
Si hay datos incrementales, crea una tabla de almacenamiento temporal, carga solo los datos incrementales en la tabla de almacenamiento temporal, copia los datos en Cloud Storage y carga los datos en la tabla de BigQuery.
Configuración de seguridad
BigQuery usa la gestión de identidades y accesos para gestionar el acceso a los recursos. Los roles predefinidos de BigQuery proporcionan acceso granular a un servicio específico y están diseñados para admitir casos prácticos y patrones de control de acceso comunes. Puedes usar roles personalizados para proporcionar un acceso aún más detallado personalizando un conjunto de permisos.
Los controles de acceso de las tablas y los conjuntos de datos especifican las operaciones que pueden realizar los usuarios, los grupos y las cuentas de servicio en las tablas, las vistas y los conjuntos de datos. Las vistas autorizadas te permiten compartir resultados de consultas con usuarios y grupos concretos sin darles acceso a los datos de origen subyacentes. Con la seguridad a nivel de fila y la seguridad a nivel de columna, puedes restringir quién puede acceder a qué filas o columnas de una tabla. La máscara de datos te permite ocultar de forma selectiva los datos de las columnas a grupos de usuarios, pero estos podrán seguir accediendo a las columnas.
Cuando aplicas controles de acceso, puedes conceder acceso a los siguientes usuarios y grupos:
- Usuario por correo electrónico: da acceso al conjunto de datos a una cuenta de Google individual.
- Agrupar por correo electrónico: da acceso al conjunto de datos a todos los miembros de un grupo de Google.
- Dominio: da acceso al conjunto de datos a todos los usuarios y grupos de un dominio de Google.
- Todos los usuarios autenticados: da acceso al conjunto de datos a todos los titulares de cuentas de Google (hace que el conjunto de datos sea público).
- Propietarios del proyecto: da acceso al conjunto de datos a todos los propietarios del proyecto.
- Lectores del proyecto: da acceso al conjunto de datos a todos los lectores del proyecto.
- Editores del proyecto: concede acceso al conjunto de datos a todos los editores del proyecto.
- Vista autorizada: da acceso de vista al conjunto de datos.
Cambios en el flujo de datos
En las siguientes secciones se explica cómo cambiar las canalizaciones de datos al migrar de Hive a BigQuery.
Sqoop
Si tu canalización usa Sqoop para importar datos en HDFS o Hive para procesarlos, modifica el trabajo para importar datos en Cloud Storage.
Si va a importar datos en HDFS, elija una de las siguientes opciones:
- Copia los archivos de salida de Sqoop en Cloud Storage con Hadoop DistCp.
- Envía los archivos directamente a Cloud Storage mediante el conector de Cloud Storage. El conector de Cloud Storage es una biblioteca Java de código abierto que te permite ejecutar tareas de Apache Hadoop o Apache Spark directamente en los datos de Cloud Storage. Para obtener más información, consulta Instalar el conector de Cloud Storage.
Si quieres que Sqoop importe datos a Hive que se ejecuta enGoogle Cloud, dirígelo directamente a la tabla de Hive y usa Cloud Storage como almacén de Hive en lugar de HDFS. Para ello, asigna la propiedad hive.metastore.warehouse.dir a un segmento de Cloud Storage.
Puedes ejecutar tu tarea de Sqoop sin gestionar un clúster de Hadoop. Para ello, usa Dataproc para enviar tareas de Sqoop e importar datos a BigQuery.
Spark SQL y HiveQL
El traductor de SQL por lotes o el traductor de SQL interactivo pueden traducir automáticamente tu Spark SQL o HiveQL a GoogleSQL.
Si no quieres migrar tu Spark SQL o HiveQL a BigQuery, puedes usar Dataproc o el conector de BigQuery con Apache Spark.
ETL de Hive
Si hay trabajos de ETL en Hive, puedes modificarlos de las siguientes formas para migrarlos desde Hive:
- Convierte la tarea de ETL de Hive en una tarea de BigQuery mediante el traductor de SQL por lotes.
- Usa Apache Spark para leer y escribir en BigQuery mediante el conector de BigQuery. Puedes usar Dataproc para ejecutar tus tareas de Spark de forma rentable con la ayuda de clústeres efímeros.
- Reescribe tus flujos de procesamiento con el SDK de Apache Beam y ejecútalos en Dataflow.
- Usa Apache Beam SQL para reescribir tus canalizaciones.
Para gestionar tu flujo de procesamiento ETL, puedes usar Cloud Composer (Apache Airflow) y plantillas de flujos de trabajo de Dataproc. Cloud Composer proporciona una herramienta para convertir flujos de trabajo de Oozie en flujos de trabajo de Cloud Composer.
Dataflow
Si quieres migrar tu flujo de procesamiento de ETL de Hive a servicios en la nube totalmente gestionados, te recomendamos que escribas tus flujos de procesamiento de datos con el SDK de Apache Beam y los ejecutes en Dataflow.
Dataflow es un servicio gestionado para ejecutar flujos de procesamiento de datos. Ejecuta programas escritos con el framework de código abierto Apache Beam. Apache Beam es un modelo de programación unificado que te permite desarrollar flujos de procesamiento por lotes y de streaming.
Si tus flujos de procesamiento de datos son de movimiento de datos estándar, puedes usar plantillas de Dataflow para crear rápidamente flujos de procesamiento de Dataflow sin escribir código. Puedes consultar esta plantilla proporcionada por Google, que te permite leer archivos de texto de Cloud Storage, aplicar transformaciones y escribir los resultados en una tabla de BigQuery.
Para simplificar aún más el procesamiento de datos, también puedes probar Beam SQL, que te permite procesar datos mediante instrucciones similares a SQL.