Nedim Badzak

Introduction In today's data-driven landscape, the efficiency and flexibility of your ETL processes can make or break your analytics strategy. With its robust, cloud-native architecture, Snowflake offers a powerful solution for building and managing data pipelines that can adapt to the ever-evolving needs of modern businesses. Unlike traditional ETL tools, Snowflake’s suite of features, ranging from Snowpipe to Dynamic Tables, enables organizations to streamline their data workflows directly within the platform, reducing the need for external tools and ensuring seamless integration. This blog explores how Snowflake’s native ETL capabilities simplify the extraction, transformation, and loading of data and provide unique advantages in scalability, cost-efficiency, and real-time data processing. Whether you're dealing with structured or semi-structured data, Snowflake's features empower you to create customized, high-performance data pipelines that are flexible and easy to manage, making it an essential tool for any data engineering team. Intro to Snowflake Snowflake is a cloud-based data platform that enables organizations to store, process, and analyze their data using a scalable and flexible architecture. Its architecture consists of three layers: storage, compute and services.  The storage layer is responsible for storing data. It's highly scalable and decoupled from other layers, allowing independent scaling and ensuring high availability through automatic replication and backup. The compute layer, or virtual warehouse layer, is where data processing occurs. It uses scalable virtual warehouses to execute queries in parallel, providing workload isolation and flexibility. Warehouses can be suspended when not in use, optimizing cost efficiency. The services layer acts as Snowflake's control plane, handling query optimization, metadata management, and security. It ensures efficient query execution, maintains schema and query history, and enforces access controls, providing a consistent and secure environment. One of the nice things about Snowflake is that it is cloud provider agnostic, meaning it can combine multiple cloud providers, such as AWS and GCP, and treat them as one as far as data retrieval goes.  Source Snowflake's key features, which we will be using for enabling continuous data pipelines, are Snowpipe, Stream, Task, and Kafka connectors. These features help to address the challenges of collecting real-time data, data change capture, scheduling, and orchestration. What is an ETL? An ETL (Extract, Transform, Load) process is a fundamental component in data warehousing and analytics, facilitating the seamless transfer of data from various sources into a target destination, typically a data warehouse or a database. Extract involves gathering data from disparate sources, Transform entails cleaning, reformatting, and enriching the extracted data to fit the target schema or requirements, and Load refers to the process of loading the transformed data into the target destination efficiently. ETL pipelines ensure data quality, consistency, and accessibility, forming the backbone of data integration and analytics workflows across industries. Lately, more and more people have started popularizing the term ELT (Extract, Load, Transform), with the main difference to ETL being that the data is transformed in the data warehouse itself rather than somewhere else and then loaded into it. This would be a good example of an ELT since the transformation is indeed happening on Snowflake directly, but for the purposes of this blog, let’s still call it ETL. Snowflake ETL Example We will be covering a pipeline that takes data from multiple sources such as Kafka and S3, saves them in Ingestion tables, creates a stage to track new data ingestions, transforms data using Stored Procedures that are scheduled using Tasks, move data to aggregate tables accordingly and save in a final table which can be queried from a UI or some other service. We can see the diagram for this pipeline in the image below: Covering E of ETL For this example, let’s take Kafka as the main source. We have multiple Kafka topics containing data and need to read data from them. One option would be to create a consumer for each topic and write the data into Snowflake using a JDBC driver. Another would be a Kafka connector, which would directly read from Kafka. The third and most convenient option (covered in this blog) would be using Kafka Connect. It enables us to provide a list of topics, automatically scan all topics for changes, keep track of the last read record, and insert everything into a fitting Snowflake table. The basics of it are that we have a Kafka Producer, which puts data into Kafka (with the best practice being to also use a Schema Registry to have “expected”, structured data), which is then read by Kafka Connect and, using a Sink Connector, put into different tables inside Snowflake. Let’s show one example of a row inserted by Kafka Connect. As you can see, we have two columns which get populated, record_metadata, that gives us info about what time the record was created, the offset, partition, and Kafka topic which it was read from. The record_metadata column contains the actual data that is read from Kafka and inserted as raw JSON. Do note that Kafka could’ve contained data in a different format than plain JSON, for example, Avro, but that gets deserialized by Kafka Connect and different plugin converters so that we always receive the same input inside Snowflake. Let’s also add a Snowpipe Stage to read additional data from S3. This step requires us to create a trust relationship between the Snowflake account and S3 bucket (read more in official documentation Automating Snowpipe for Amazon S3), and once done, it can be as simple as doing: CREATE STORAGE INTEGRATION s3_int TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = 'S3' ENABLED = TRUE STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::001234567890:role/myrole' STORAGE_ALLOWED_LOCATIONS = ('*') STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket1/mypath1/sensitivedata/', 's3://mybucket2/mypath2/sensitivedata/'); CREATE STAGE mystage URL = 's3://mybucket/load/files' STORAGE_INTEGRATION = s3_int; Once we want to read data from S3, we can query the stage just like we would query a regular table with SQL, and it will read directly from S3 and give us an output that resembles a queried table output.  Now that we have multiple sources to read data from, let’s try querying it. Since all of our data is currently in JSON format, and that is native to Snowflake with it being called VARIANT, we can query it like this: SELECT record_content[‘id’] FROM mytable; In larger JSONs, this can be significantly slower than querying a regular column. Regarding Extraction, we could create a pipe that would load the data automatically from a Stage to a Snowflake table: CREATE PIPE snowpipe_db.public.mypipe AUTO_INGEST = TRUE AS COPY INTO snowpipe_db.public.mytable FROM @snowpipe_db.public.mystage FILE_FORMAT = (type = 'JSON'); Okay, we extracted the data. Next, let’s work on Transforming and Loading it. Covering T of ETL Okay, we have extracted and consolidated the data from different sources into Snowflake. Now, let’s work on transforming the data using different Snowflake tools and techniques, such as Snowflake Tasks and Streams. Another possibility could be using Dynamic Tables as an alternative for tasks and streams. Using Tasks and Streams in Snowflake can offer more control and flexibility compared to Dynamic Tables, particularly for complex data workflows. They allow you to define and schedule specific actions, like incremental data processing, in a modular way. This means you can tailor the pipeline to suit specific business needs, trigger updates based on events, and integrate with external systems as needed. While Dynamic Tables automate some of this process, they might lack the granular control that Tasks and Streams provide, making the latter a better choice for highly customized data operations, so we will explore them in more detail. What is a Task in Snowflake? Tasks in Snowflake are versatile tools that can be used to automate various types of operations. Tasks can handle it all, whether it's a single SQL statement, a call to a stored procedure, or procedural logic using Snowflake Scripting. When combined with table streams, tasks can form continuous ETL workflows that process recent changes in the table with exactly-once semantics for inserted, updated, or deleted data. You can create tasks in Snowflake via SQL, which can be triggered based on conditions, events, or scheduled intervals. Below is a blueprint for creating a task in Snowflake SQL: CREATE OR REPLACE TASK db.schema.task_name WAREHOUSE = my_warehouse SCHEDULE = '1 MINUTE' SUSPEND_TASK_AFTER_NUM_FAILURES = 10 WHEN SYSTEM$STREAM_HAS_DATA('db.schema.streamA') AS INSERT INTO db.schema.tableB ( SELECT tableA.name, tableB.id as primary_id, tableB.category, CASE WHEN tableA.count > 50 THEN 1 ELSE 0 END AS enough FROM tableA LEFT JOIN tableB on tableA.id = tableB.id ); Alternatively, you can create a task that calls a stored procedure: CREATE OR REPLACE TASK db.schema.task_name WAREHOUSE = my_warehouse SCHEDULE = '1 MINUTE' SUSPEND_TASK_AFTER_NUM_FAILURES = 10 WHEN SYSTEM$STREAM_HAS_DATA('db.schema.streamA') AS CALL db.schema.my_PROCEDURE(); Streams in Snowflake Streams are integral to Snowflake's ETL capabilities. They track changes to a table - new, updated, or deleted rows, allowing you to process only the modified data, which is crucial for efficient ETL workflows. This targeted approach reduces the need to process the entire dataset repeatedly, saving time and computational resources. Additionally, streams play a critical role in maintaining data consistency, ensuring that your ETL processes are always working with the most up-to-date information. This helps prevent the risk of processing stale or outdated data, ultimately leading to more accurate and reliable analytics outcomes. Here is an example of a stream definition: CREATE OR REPLACE STREAM db.schema.streamA ON TABLE db.schema.tableA; Streams, combined with tasks, enable the construction of a flexible and efficient ETL pipeline. This pipeline adapts to real-time data changes, automates data workflows, and provides a foundation for continuous improvement in data processing capabilities. Using Tasks and Streams Together By querying a stream, you can obtain the delta of changes and create a view that fetches and transforms this data. For example: CREATE OR REPLACE VIEW db.schema.viewA AS SELECT * FROM db.schema.streamA; You can then create a task that reads from this view and writes the transformed data into a new table, aggregating it as needed: CREATE OR REPLACE TASK db.schema.task_name WAREHOUSE = my_warehouse SCHEDULE = '1 MINUTE' WHEN SYSTEM$STREAM_HAS_DATA('db.schema.streamA') AS INSERT INTO db.schema.new_tableA SELECT * FROM db.schema.streamA; This task can be scheduled based on time, trigger, and after another task execution, providing a robust mechanism for managing your ETL workflows in Snowflake. In the code snippet above, we are checking if stream db.schema.streamA has new data available every 1 minute, and if it does, we are inserting that delta of data from the stream into new_tableA. Covering L in Snowflake Finally, let’s talk about loading the data from newly aggregated tables to target tables. We will go through different topics, such as the purpose of the loading phase, how upserts are done on Snowflake, how we can leverage Snowpipe for loading purposes etc. Finally, we will touch base on automating the loading process.  The Purpose of the Loading Phase In the ETL process, the final step is loading the transformed data into Snowflake’s target tables. This step ensures that the data is organized and ready for analysis, reporting, or consumption by various applications and services. Organizing Data for Consumption After transforming your data to fit the desired schema, the next step is to load it into final tables that your analytics tools, dashboards, and other applications will consume. This is where Snowflake's flexible data warehousing capabilities shine. Loading Data into Final Tables To load data into Snowflake’s target tables, you can use several methods, depending on your needs. For simple transformations, we can use simple INSERT statements to move data from staging tables to final tables: INSERT INTO db.schema.final_table SELECT * FROM db.schema.transformed_table; For cases where we need to handle upserts (updates and inserts), MERGE statements are effective. They allow you to insert new records and update existing ones in a single operation without having to do some complex logic while still being fast and efficient. MERGE INTO db.schema.final_table AS target USING db.schema.transformed_table AS source ON target.id = source.id WHEN MATCHED THEN UPDATE SET target.column1 = source.column1, target.column2 = source.column2 WHEN NOT MATCHED THEN INSERT (id, column1, column2) VALUES (source.id, source.column1, source.column2); We previously mentioned Snowpipe as something related to Extraction which can be used to load data into Snowflake, but it can also be used in the Load part of the process as it has capabilities to automatically load data as soon as it is available in a specified stage, which could be created on a transformed table, not just source tables. Finally, besides being able to import data from the outside easily, Snowflake can also export large datasets to an external location. For large datasets, bulk loading methods using COPY INTO commands can efficiently load data from external files into Snowflake tables or Snowflake tables to external files: COPY INTO 's3://your-bucket-name/path/to/export/file_prefix' FROM (SELECT * FROM your_table_name) CREDENTIALS = ( AWS_KEY_ID = 'your_aws_access_key_id' AWS_SECRET_KEY = 'your_aws_secret_access_key' ) FILE_FORMAT = (TYPE = CSV, FIELD_OPTIONALLY_ENCLOSED_BY = '"', COMPRESSION = NONE) OVERWRITE = TRUE; Automating the Loading Process You can use Snowflake tasks to schedule and manage the loading operations to automate the loading process and ensure data is consistently updated. CREATE OR REPLACE TASK db.schema.load_task WAREHOUSE = my_warehouse SCHEDULE = '1 HOUR' AS MERGE INTO db.schema.final_table AS target USING db.schema.transformed_table AS source ON target.id = source.id WHEN MATCHED THEN UPDATE SET target.column1 = source.column1, target.column2 = source.column2 WHEN NOT MATCHED THEN INSERT (id, column1, column2) VALUES (source.id, source.column1, source.column2); Making Data Available for Consumption Once the data is loaded into the final tables, it is ready to be consumed by various BI tools, dashboards, and applications. Snowflake’s ability to handle structured and semi-structured data makes it a powerful platform for diverse analytics needs, ensuring that the loaded data is always available, scalable, and performant. Final Thoughts on ETL for Snowflake To fully comprehend the role of ETL within Snowflake, it’s essential to understand the platform’s built-in capabilities for constructing data pipelines, which we’ve explored above. This understanding highlights Snowflake’s strengths and reveals areas where dedicated ETL tools might still be necessary. There are additional features and concepts we haven’t covered that could be valuable depending on your specific use cases: Time Travel and Fail-safe: Provides access to historical data and additional safeguards, allowing you to recover previous data states and protect against accidental data loss or corruption. Data Sharing: Enables secure sharing of data across different Snowflake accounts without the need to move the data, facilitating collaboration. Materialized Views: Enhances data retrieval by automatically maintaining and storing query results, which speeds up the performance of complex, frequently run queries. Snowflake’s Data Marketplace: Enriches data pipelines with external sources, offering seamless integration of a wide range of datasets and services into your analytics workflows. These features underscore Snowflake’s effectiveness as a data engineering solution, yet the question remains: why might an external ETL tool still be necessary? Deciding whether to use a separate ETL tool despite Snowflake’s comprehensive native capabilities is complex. While Snowflake’s features are robust and particularly useful for certain data management tasks, there are several reasons why an external ETL tool could be beneficial: Real or Near-Real Time Processing: For critical real-time data processing, some ETL tools are better suited to handle streaming data more effectively than traditional data warehousing methods. Cost and Performance Optimization: Performing complex transformations within Snowflake can be resource-intensive and costly, as Snowflake charges based on warehouse usage per minute. Offloading heavy computational tasks to an ETL tool can help manage costs and improve performance. Scalability and Flexibility: For organizations managing large data volumes, dedicated ETL tools might offer greater scalability or be optimized for large-scale operations, even though Snowflake does provide a range of warehouses tailored to different needs. While Snowflake’s native tools are powerful and can meet many use cases, building data pipelines using Snowflake alone can be akin to creating a custom data platform. This approach requires a modular assembly of various components like Snowpipe, Streams & Tasks, and Materialized Views, necessitating significant integration work to achieve a cohesive data management process. Conclusion Snowflake is a powerful and flexible cloud-based data platform offering robust data storage, processing, and analysis capabilities. Its architecture integrates storage, compute, and services layers, and its cloud-provider agnostic nature allows seamless data accessibility across multiple clouds. Snowflake’s features, such as Snowpipe, Streams, and Tasks, alongside Kafka connectors, enable the creation of continuous data pipelines, facilitating real-time data ingestion, transformation, and loading. This ensures that data from various sources is accurately, efficiently processed, and made available for analysis and consumption. Embracing Snowflake’s capabilities allows organizations to build efficient, scalable, cost-effective data management solutions.   (more…)

Orchestrating complex workflows and managing data dependencies between tasks can be a challenging and time-consuming task. Something like this would take a lot of time and resources to build and could very well be a big hassle to maintain. That’s where AWS Step Functions come to the rescue. AWS Step Functions provide a low-code visual workflow service that simplifies this process. In this blog post, we will explore how to achieve partially dependent parallel flows in AWS Step Functions, enabling you to pass data between parallel states during their runtime efficiently. AWS Step Functions AWS Step Functions is a low-code powerful workflow service that offers a visual representation of application workflows, integration with various AWS services, and real-time error detection.  By leveraging Step Functions, you can create workflows with a sequence of steps where the output of one step becomes the input for the next. With over 220 AWS service integrations available through AWS SDK integration tasks, you can call AWS SDK actions directly without writing additional code. Additionally, Step Functions can be triggered on specific events or timers using EventBridge Scheduler, adding flexibility and automation to your workflows. Parallel Workflows One feature that could especially be useful is Parallel Workflows. Parallel Workflows provide a way to run multiple steps that run concurrently and can, if wanted, wait for all tasks to complete before continuing. You can also choose which output you want at the end of the parallel execution. This all sounds great, but a certain caveat might not be noticeable at first. What if you want to pass data between parallel states during their runtime? The challenge Let’s first talk about why one would want to pass data between parallel states at runtime. For example, let’s imagine that you have the following situation: I want to note that Glue Jobs and Step functions are used as an example here, any Step Function supported service could be used instead of them, both here and in the rest of the text. If the two-step functions need all three Glue jobs to complete, you could put these step function runs at the end of the parallel workflow and then run them, parallel again. But what if one of the Glue jobs has a very short duration and the other one has a rather long one? It wouldn’t be great if you had to wait for all of them to complete, only to have the possibly lengthy sub-step functions run when one of them could’ve been run a long time ago. This may seem like it’s not a big deal at first, but as with anything, when put at scale, this could turn out to be a major bottleneck. Wouldn’t it be great if you could say, I want Step Function 1 to be dependent on Glue job 1 and Glue job 3, but not Glue job 2, or Step Function 2 to be dependent on Glue job 2 and Glue job 3, but not Glue job 1? The first thing that we want to address is: What if I could tell step function 1 to wait for both Glue job 1 and Glue job 3 executions to be finished? Well, Step functions don’t support this sort of dependency relationship, so we need to create our own workaround for this. Achieving dependency in parallel flows The first thing we need to create is a poller that will periodically check for Glue job 3 completion.  Even though this may seem overwhelming initially, it is fairly simple. We have a timer which calls Get Job Status, which then checks if the desired job is completed or not, and depending on the outcome, either starts the step function or returns to the timer and calls Get Job Status again. If Glue job 1 finishes before Glue job 2, we are going to wait for Glue job 2 to finish. On the other hand, if Glue job 2 finishes before Glue job 1, Get Job run will be run only once and there will be no need for the polling procedure.  This poller approach is relatively common in step functions, as you often need to do a certain action periodically. It is even covered in the official documentation of AWS. Now the only thing that remains is how do we check for completion of the Glue job, or better yet, what should our Get Job Status be? Well, if we look at the documentation, it says that Get Job Status should be an API call, and what better way to do fast repeatable API calls than Lambda functions?  Okay, now all that’s left to do is write the actual Lambda function. We first need to import the boto3 client and pass along the desired job name. We can then use GetJobRuns to get all runs of that Glue Job. This could present an issue as we can’t always be sure if the latest successful run is from this step function run or if the Glue job hasn’t yet been started at all, which could easily be the case if we had another Glue Job before Glue Job 3. The solution for this is to pass along the Step Function started on timestamp, as well as the frequency at which this Step function is being run, by, for example, an Eventbridge scheduler. Taking all this into consideration, we end up with: import json import boto3 from datetime import datetime, timedelta sfnDateFormat = '%Y-%m-%dT%H:%M:%S.%f%z' def isWithinLatestSFNRun(sfnStartedOn: datetime, jobStartedOn: datetime, triggerOffset: int) -> bool: rounded = sfnStartedOn - timedelta( minutes=sfnStartedOn.minute % triggerOffset, seconds=sfnStartedOn.second, microseconds=sfnStartedOn.microsecond ) print(f"Rounded: {rounded}") return jobStartedOn >= rounded def lambda_handler(event, context): client = boto3.client('glue') attempt = event['attempt'] + 1 jobName = event['job']['name'] sfnStartedOn = datetime.strptime(event['sfn']['startedOn'], sfnDateFormat) sfnTriggerOffset = event['sfn']['triggerOffset'] print(f"SFN Started on: {sfnStartedOn} \n SFN Trigger Offset: {sfnTriggerOffset}") paginator = client.get_paginator('get_job_runs') response_iterator = paginator.paginate( JobName=jobName, PaginationConfig={ 'MaxItems': 3 } ) for jobRuns in response_iterator: for jobRun in jobRuns['JobRuns']: StartedOn, JobRunState = jobRun['StartedOn'], jobRun['JobRunState'] if isWithinLatestSFNRun(sfnStartedOn, StartedOn, sfnTriggerOffset) and JobRunState == 'SUCCEEDED': return { 'attempt': attempt, 'statusCode': 200, 'status': 'DONE' } return { 'attempt': attempt, 'statusCode': 200, 'status': 'WAITING' } Conclusion Every approach comes with a certain drawback, and this isn’t any different. The most obvious one would be pulling all Glue job runs instead of the specific, desired one. Boto3 supports GetJobRun for Glue jobs, and you need to pass the ID of the job run to get information about that run. But, we can’t know the Glue job run ID as it is generated at Glue job run start, and we would need to pass that data at runtime to the poller, which isn’t trivial and could produce unexpected dependency issues when one Glue job starts before the other.. One might think that a good idea would be to somehow set the Glue job run ID manually. Even though this would be a great approach, Glue jobs don’t, unfortunately, support setting their run ID manually, even though they take it as a parameter. Upon further research, you’ll find it is only used to provide a previous job run ID in the retry phase, as explained in their documentation. As you can see above, one solution for pulling all glue job runs is using a paginator in combination with MaxItems. By limiting the retrieval to a specified number of the latest Glue job runs, we can perform checks on a significantly reduced dataset. Although it may appear as a minor enhancement initially, this approach greatly improves performance and execution times. Since Lambda functions are designed for fast and frequent invocation with minimal execution times, optimizing data retrieval contributes to overall efficiency. With the ability to pass along data in partially dependent parallel flows in AWS Step functions, the possibilities for improving your workflow are endless. An example step function which was previously running for more than an hour has now been reduced to under 30 minutes, which showcases how seemingly minor optimizations like this can go a long way in improving execution times at large, and while achieving partially dependent parallel flows in AWS Step Functions does require a workaround, the benefits of optimized workflow execution times make it worthwhile. "How to achieve partially dependent parallel flows in AWS Step Functions" Tech Bite was brought to you by Nedim Badžak, Junior Software Engineer at Atlantbh. (more…)