Programa
Snowflake Foundations
0 min
Las canalizaciones de datos son un componente central de la ingeniería de datos. Representan la serie de pasos que trasladan los datos brutos de los sistemas fuente a un almacén de datos, donde pueden limpiarse, transformarse y prepararse para el análisis. Un pipeline bien diseñado garantiza que los datos fluyan de forma eficaz, fiable y segura desde la ingesta hasta la comprensión.
Snowflake, una potente plataforma de datos basada en la nube, es especialmente adecuada para construir canalizaciones de datos escalables, flexibles y rentables. Con funciones integradas de carga, transformación, automatización y supervisión de datos, Snowflake permite a los equipos de datos simplificar los flujos de trabajo complejos y centrarse en aportar valor a partir de los datos.
En este tutorial, recorreremos el proceso de creación de una canalización de datos en Snowflake, desde la configuración de tu entorno hasta la automatización de las transformaciones de datos mediante herramientas nativas como Streams y Tareas. Parte de esto requerirá conocer el proceso de integración de un servicio en la nube y los sistemas de gestión de eventos. Para ello, te recomiendo que consultes la documentación de tu servicio en la nube concreto.
Si acabas de empezar con Snowflake, considera nuestra curso de conocimientos básicos de Snowflake para ponerte manos a la obra y aprender a consultar, modelar y gestionar datos.
Comprender la arquitectura de Snowflake
La arquitectura única de Snowflake está bien diseñada para soportar canalizaciones de datos. Snowflake separa el cálculo y el almacenamiento, lo que significa que puedes escalar cada uno independientemente.
Los datos se almacenan de forma centralizada en la capa de almacenamiento de la nube, mientras que el cálculo se realiza en almacenes virtuales, que pueden iniciarse, detenerse y redimensionarse según sea necesario.
Esta arquitectura permite que varios usuarios y procesos accedan a los mismos datos sin contención, lo que la hace ideal para cargas de trabajo de canalización que implican ingestión, transformación y análisis.
También añade una capa de seguridad en la que los cambios en los datos pueden producirse sin afectar directamente a la capacidad de un usuario para consultar esos datos.
Para obtener más información y conocer la capa de la nube de la que no se habla aquí, te recomiendo esta inmersión profunda en la arquitectura de Snowflake.
Componentes clave de las canalizaciones Snowflake
Dentro de cada tubería, debemos considerar los pasos necesarios. Generalmente, escenificarás los datos antes de almacenarlos en un almacén virtual. Este almacén virtual contendrá la base de datos que almacena las tablas y su esquema. He aquí las definiciones de estos elementos esenciales utilizados en las canalizaciones de datos Snowflake:
- Etapas: Almacenes temporales o externos utilizados para cargar/descargar datos.
- Almacenes virtuales: Recursos informáticos utilizados para procesar consultas SQL y operaciones con datos.
- Bases de datos: Contenedores lógicos para organizar tus datos. Estos suelen almacenar una colección tablas
- Esquemas: Subdivisiones dentro de una base de datos para agrupar tablas y objetos relacionados. Define cómo se relacionan las tablas entre sí.
- Tablas: Dónde se almacenan los datos estructurados
Configurar tu entorno Snowflake
Vamos a asegurarnos de que tienes un entorno Snowflake listo para funcionar. Si eres nuevo en Snowflake, sigue estos pasos para empezar:
Crear una cuenta Snowflake
Si no tienes una cuenta Snowflake, hay una prueba gratuita de 30 días que incluye (en el momento de escribir esto) 400 $ de uso gratuito. ¡Esto debería bastarnos para probar unos cuantos pipelines de muestra!
- Visita Prueba de Copo de Nieve y regístrate para obtener una cuenta gratuita.
Formulario de inscripción para la Prueba de los Copos de Nieve
- ¡Para nuestros propósitos, seleccionaremos la edición Estándar en lugar de la edición Enterprise para ahorrar en costes de crédito en el futuro si sigues utilizando Snowflake!
- Elige tu proveedor de nube preferido (AWS, Azure o GCP). Los ejemplos de este tutorial utilizarán AWS.
- El centro de datos elegido automáticamente debe ser el más cercano a ti geográficamente y debe proporcionar la mejor experiencia.
- Después de crear la cuenta, inicia sesión en la Snowflake Web UI.
Navegar por la IU
Una vez dentro, verás
- Hoja de trabajo: Donde puedes ejecutar consultas SQL.
- Bases de datos: Visualiza y gestiona bases de datos y esquemas.
- Almacenes: Monitoriza y controla los recursos informáticos.
- Historia: Haz un seguimiento de las consultas y tareas ejecutadas anteriormente
Si quieres más consejos sobre el uso de Snowflake, sigue esto Fundamentos de Snowflake curso.
Crear los recursos necesarios
Este paso cubrirá la creación de un almacén virtual, una base de datos y un esquema. Ejecuta el siguiente SQL en tu hoja de cálculo para configurar tu entorno:
-- set the user role appropriately
USE ROLE accountadmin;
-- Create a virtual warehouse
CREATE WAREHOUSE datacamp_pipeline_wh WITH WAREHOUSE_SIZE = 'XSMALL' AUTO_SUSPEND = 60;
-- Set our active working warehouse
USE WAREHOUSE datacamp_pipeline_wh;
-- create our Datacamp Sample Database
CREATE OR REPLACE DATABASE datacamp_sample_db;
-- create the Datacamp (Sample) Raw Schema
CREATE OR REPLACE SCHEMA datacamp_sample_db.raw_schema;Cargar datos en Snowflake
Como comentamos en nuestro artículo sobre La ingestión de datos en SnowflakeSnowflake admite tanto la carga por lotes (a granel) como la carga continua. La carga masiva está diseñada para cargar grandes cantidades de datos en tu tabla a la vez. Por lo general, esto se hace a intervalos programados, o se automatiza con la gestión basada en eventos de Snowpipe.
La carga continua (stream) puede realizarse utilizando cosas como Snowpipe Stream. Para ingerir datos, debemos crear integraciones de datos (etapas) con nuestro proveedor de la nube y cargar los datos utilizando COPY INTO.
Creación del escenario S3 e integración
La ingesta de datos es el primer paso de cualquier canalización. En este paso, cubriremos los aspectos básicos de la creación de una integración de almacenamiento y la creación de un escenario. Para ello, debemos imaginar que ya hemos conectado tu Snowflake a AWS con un usuario IAM de Snowflake. Si no es así, consulta la Guía de configuración de Snowflake AWS en la documentación.
-- set Role Context
USE ROLE accountadmin;
-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;
-- this creates the storage integration
-- The STORAGE_AWS_ROLE_ARN will come from the AWS configuration guide steps
-- We will allow access to all locations and block any with sensitive data
CREATE STORAGE INTEGRATION datacamp_aws
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = 'S3'
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '<iam_role>'
STORAGE_ALLOWED_LOCATIONS = ('*')
STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);
-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format
TYPE = 'CSV'
FIELD_OPTIONALLY_ENCLOSED_BY = '"';
USE SCHEMA datacamp_sample_db.raw_schema;
--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
STORAGE_INTEGRATION = datacamp_aws
URL = 's3://bucket1/path1/'
FILE_FORMAT = my_csv_format;Al final, deberías haber creado con éxito tu integración Snowflake con AWS y creado el escenario. Si quieres repasar los conceptos básicos de AWS, el siguiente artículo es una gran introducción a AWS.
Carga masiva con COPY INTO
El comando COPY INTO es el método habitual para cargar datos en Snowflake. Podemos utilizar un archivo cargado localmente en Snowflake, o utilizar nuestras conexiones en la nube. En el siguiente ejemplo, utilizaremos nuestra etapa para copiar un archivo CSV.
-- Load data using COPY INTO
-- The following will attempt to load all file in the bucket
-- To specify the particular files you can use the FILES or PATTERN settings
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @my_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Automatizar la carga con Snowpipe
Cargar manualmente cada vez que recibimos un nuevo archivo es ineficaz y lleva mucho tiempo a medida que aumenta la velocidad. En su lugar, podemos utilizar Snowpipe para cargar archivos desde nuestro escenario. A continuación, combinamos nuestras notificaciones de eventos de AWS con nuestro usuario IAM de Snowflake para activar el Snowpipe cada vez.
-- Let’ s first create our Snowpipe
CREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');A continuación, podemos utilizar algo como Amazon SQS para enviar notificaciones a nuestro usuario IAM cada vez que nuestro bucket de destino reciba un archivo.
Por último, podemos monitorizar nuestro Snowpipe utilizando:
SELECT * FROM INFORMATION_SCHEMA.PIPE_USAGE_HISTORY;¡Ahora ya no tienes que preocuparte de subir datos cada vez que llegue un archivo a tu bucket de S3! Estas notificaciones en la nube harán que tu Snowpipe cargue los datos.
Transformar datos con flujos y tareas
Los flujos y las tareas de Snowflake son herramientas integradas para seguir y transformar los datos cambiantes. Las secuencias se centran en el seguimiento de los cambios en las tablas, mientras que las tareas se utilizan para automatizar la ejecución de sentencias SQL.
Introducción a las corrientes
Un Flujo captura información de Captura de Datos de Cambios (CDC) -es decir, inserciones, actualizaciones y eliminaciones- en una tabla desde la última vez que se consultó. Suelen adoptar la forma de consultas del Lenguaje de Manipulación de Datos (LMD). Esto nos permite ver los cambios de una tabla entre dos tiempos de consulta diferentes.
-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Luego puedes consultar el flujo para obtener sólo los cambios:
SELECT * FROM my_stream WHERE METADATA$ACTION = 'INSERT';Introducción a las tareas
Las tareas te permiten automatizar la ejecución de SQL según una programación o un evento desencadenante. Las tareas pueden recurrir a SQL, Javascript, Python, etc., para automatizar procedimientos en nuestras tablas y canalizaciones.
-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
WAREHOUSE = datacamp_sample_db
SCHEDULE = '5 MINUTE'
AS
INSERT INTO raw_schema.aws_processed_data
SELECT * FROM my_stream WHERE location = ‘USA’;
-- This will allow us to test the task
EXECUTE TASK process_changes_task;Combinar flujos y tareas
Esta es la base de las cadenas de transformación automatizadas. Como has visto antes, las Tareas pueden utilizar Secuencias para introducir nuevos datos en una tabla diferente. Un proceso general puede ser como el siguiente
- Crear un Flujo para controlar los cambios en una tabla de datos
- Crea una Tarea que extraiga datos de ese flujo de forma programada
- Con cada extracción de datos, transforma los datos al formato que deseemos
- Carga esos datos en nuestra tabla de destino
De este modo, hemos creado partes de una canalización de datos. Combinando Streams y Tareas, podemos realizar transformaciones sólo en los datos recién cargados. Con sólo Tareas, potencialmente tendríamos que procesar la totalidad de nuestras tablas sin procesar cada vez que quisiéramos nuevos datos, lo que puede resultar costoso y un uso ineficiente de la potencia de cálculo.
Construir una canalización de datos de extremo a extremo
Recorramos un caso de uso sencillo de una canalización: ingesta de datos de pedidos, seguimiento de los cambios y transformación para análisis. Ya que hemos visto cómo ejecutar cada paso de forma independiente, en esta sección veremos cómo podrían ser los pasos juntos.
Paso 1: Crear integración y escenario de almacenamiento en la nube
En primer lugar, tenemos que conectar nuestro ecosistema Snowflake a nuestro almacenamiento en la nube. Utilizamos AWS como ejemplo. En Snowflake, utilizamos los comandos CREATE STORAGE INTEGRATION y CREATE STAGE para construir la conexión entre nuestro AWS y nuestra base de datos.
-- set Role Context
USE ROLE accountadmin;
-- set Warehouse Context
USE WAREHOUSE datacamp_pipeline_wh;
-- this creates the storage integration
CREATE STORAGE INTEGRATION datacamp_aws
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = 'S3'
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '<iam_role>'
STORAGE_ALLOWED_LOCATIONS = ('*')
STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/private/path/’);
-- Create file format, this is referenced by your stage to easily determine presets for your files
CREATE FILE FORMAT my_csv_format
TYPE = 'CSV'
FIELD_OPTIONALLY_ENCLOSED_BY = '"';
USE SCHEMA datacamp_sample_db.raw_schema;
--This stage is what ultimately connects your AWS files to your database schema
CREATE STAGE datacamp_s3_stage
STORAGE_INTEGRATION = datacamp_aws
URL = 's3://bucket1/path1/'
FILE_FORMAT = my_csv_format;Paso 2: Ingesta de datos con Snowpipe
En este segundo paso, creamos un Snowpipe que ejecutará el comando COPY INTO en nuestras tablas de datos brutos.
-- Let’ s first create our SnowpipeCREATE PIPE my_pipe AS
COPY INTO datacamp_sample_db.raw_schema.aws_first_upload
FROM @datacamp_s3_stage
FILE_FORMAT = (FORMAT_NAME = 'my_csv_format');Paso 3: Utiliza Streams para capturar cambios
A continuación, utilizamos un flujo en esta tabla de datos brutos para seguir los cambios.
-- Create a stream on a table
CREATE OR REPLACE STREAM my_stream ON TABLE datacamp_sample_db.raw_schema.aws_first_upload;Paso 4: Automatizar transformaciones con Tareas
A continuación, creamos una Tarea que transforma y carga estos datos recién cargados en nuestra versión procesada de la tabla. Esto se activará en un horario para comprobar si hay nuevos datos. Si no hay datos nuevos, la Tarea no se ejecutará.
-- Create a task to process changes and filter data
-- We want changes that occur in the US, for example
-- Schedule to run this task every 5 minutes from our stage
CREATE OR REPLACE TASK process_changes_task
WAREHOUSE = datacamp_sample_db
SCHEDULE = '5 MINUTE'
AS
INSERT INTO raw_schema.aws_processed_data
SELECT * FROM my_stream WHERE location = ‘USA’;
-- This will allow us to test the task
EXECUTE TASK process_changes_task;Paso 5: Supervisar y gestionar
Por último, queremos asegurarnos de que nuestros conductos funcionan según lo previsto. Asegúrate de utilizar los comandos SHOW PIPES y SHOW TASKS para verificar que están correctamente configurados y en funcionamiento. A continuación, utiliza las tablas de metadatos de INFORMATION_SCHEMA para hacer un seguimiento del rendimiento y los errores de la carga de datos.
Buenas prácticas para el desarrollo de canalizaciones de datos en Snowflake
Como siempre, queremos asegurarnos de que estamos operando con las mejores prácticas en nuestro desarrollo. Lo que hay que tener en cuenta es optimizar el rendimiento y el coste, garantizar la calidad y fiabilidad de los datos e implantar los niveles adecuados de seguridad y acceso.
Optimiza el rendimiento y el coste
Nuestro objetivo debe ser operar siempre nuestros oleoductos con la mayor eficacia posible y el menor coste. De este modo, minimizamos nuestro impacto en los servidores y proporcionamos los datos de la forma más puntual posible. Aquí tienes algunos consejos:
- Utiliza la suspensión automática en los almacenes para evitar cargos por inactividad. Esto reduce los costes de computación al tener almacenes operativos, ya que cobran por tiempo.
- Carga archivos comprimidos y ponlos por lotes en tamaños óptimos. Los archivos demasiado pequeños consumen demasiada sobrecarga de red al necesitar generar conexiones para cargar cada archivo. Los archivos demasiado grandes pueden ocupar demasiado ancho de banda y arriesgarse a fallar. Snowflake recomienda deltas de 100-250 MB de tamaño.
- Programa las transformaciones durante las ventanas de bajo uso. Esto significa que no compites por los recursos.
Garantizar la calidad y fiabilidad de los datos
La calidad y la fiabilidad de los datos son aspectos críticos que todo ingeniero de datos debe tener en cuenta. Quieres asegurarte de que los datos que se cargan son utilizables sin errores. También debes asegurarte de que tu canalización de datos no provocará inadvertidamente la caída de datos cuando haya problemas. He aquí algunos consejos para la gestión de la calidad de los datos:
- Valida los datos antes y después de la ingestión. Esto puede hacerse con comprobaciones periódicas de las tablas de etapas y alertas.
- Utiliza secuencias y tablas de preparación para aislar y arreglar los registros problemáticos. Como sólo se introducen los datos modificados, podemos minimizar el alcance de nuestros problemas utilizando flujos.
- Aplica restricciones únicas o lógica de deduplicación cuando sea necesario. Esto puede evitar cosas como
NULLdatos donde no podemos tener ninguno o pre-deduplicar en caso de solapamiento de tiradas de datos.
Implantar controles de seguridad y acceso
Por último, es importante garantizar un acceso y una seguridad adecuados a tus datos. Algunas tablas pueden contener información sensible o no deben ser alteradas salvo por un grupo selecto. Sigue estos consejos para gestionar adecuadamente tu seguridad y acceso en Snowflake:
- Utiliza funciones y subvenciones para restringir el acceso.
- Protege los datos sensibles con enmascaramiento a nivel de columna o políticas de acceso a filas.
- Rota las credenciales de almacenamiento y controla el uso del escenario.
Conclusión
Construir canalizaciones de datos en Snowflake es un proceso ágil y escalable, perfecto para los principiantes que se adentran en el mundo de la ingeniería de datos. Con su arquitectura intuitiva, sus sólidas opciones de ingesta y sus herramientas de automatización integradas, como Streams y Tareas, Snowflake permite a los equipos de datos pasar rápidamente de los datos brutos a la información procesable.
Siguiendo esta guía, dispondrás de los conocimientos básicos para configurar tu propia canalización Snowflake: ingestión, transformación y orquestación, todo dentro de una única plataforma. Para más información sobre Snowflake, consulta las siguientes guías:
