Building ETL Pipeline in Snowflake

Introduction At the time of writing this blog, it seems the most popular topics in software engineering are Artificial Intelligence (AI) and Large Language Models (LLMs). LLMs are being proactively integrated into production systems for various purposes and goals, leading to the critical question: “How can we be certain the model performs well?” Traditional software is deterministic and can be tested with clear pass/fail decisions. On the other hand, LLMs are notorious for being non-deterministic, which is embedded in their very nature. Without getting into details on the underlying architecture or how they perceive the world, LLMs are known for their hallucinations, drifting away from their targeted task, being too agreeable, etc.  These types of problems force us to adopt a different approach when it comes to grading the quality of our LLM-dependent systems. We shift our perspective to evaluation, i.e., scoring the output by a predefined scale, usually [0,1], instead of the traditional binary pass/fail. Different frameworks try to provide the best solution, and the one that stands out amongst many is DeepEval. What is DeepEval? DeepEval is an open-source Python evaluation framework for LLMs. With the main goal being to unit test LLMs, DeepEval operates similarly to Pytest, along with the additional caveats attributed to LLMs. DeepEval supports end-to-end as well as component-level evaluation.  DeepEval’s approach to evaluation relies on the following components: Test Case - a single interaction between the user and the LLM. This is a container that includes data for a specific evaluation and consists of: input - the prompt sent to the LLM actual_output - the recorded LLM response expected_output - the gold standard response for comparison (if available) Metrics - definition of the scoring rule used to evaluate the Test Case. They may be: Deterministic - use fixed algorithms and formulae (ex., exact matches, regex, ROUGE/BLEU, LaBSE...) Non-deterministic (LLM-as-a-Judge) - use an independent powerful model to “read” and assess the output. Evaluation Dataset - collection of Test Cases grouped together Individual Test Case Evaluation Ideal evaluation workflow using DeepEval Basic evaluation through an example The problem We can imagine working on an LLM-based assistant that answers questions from retrieved documents. How do we verify whether the response is faithful to the source and is semantically correct? Setup pip install deepeval sentence-transformersexport OPENAI_API_KEY="your-api-key-here" The OpenAI key is needed for the LLM-as-a-judge metric, while deterministic metrics run without external calls.  The test case from deepeval.test_case import LLMTestCasetest_case = LLMTestCase( input="What is Varnish?", actual_output="Varnish is an HTTP reverse proxy that caches responses in RAM.", expected_output="Varnish is a caching HTTP reverse proxy that accelerates web applications.", retrieval_context=["Varnish is a HTTP reverse proxy and a caching server that ","speeds up web applications by caching HTTP responses." ]) Two metrics will be used for this test case. One is an LLM-as-a-Judge metric, and the other is a deterministic metric that will utilize LaBSE. The sentence-transformers package is used for LaBSE.  The retrieval_context is text fetched from a knowledge base that the LLM uses as source material when generating its answer. The retrieval_context is needed when we want to test the model against specific issues. (hallucinations, etc.) Language-agnostic BERT Sentence Embedding (LaBSE) is a multilingual sentence embedding model. It is used to compute the cosine similarity between the actual and expected outputs to evaluate semantic similarity. Metric 1: Faithfulness (LLM-as-a-Judge) LLM-as-a-Judge metrics use a secondary independent LLM to evaluate and score the primary model’s output. DeepEval has different built-in options: from deepeval.metrics import ( AnswerRelevancyMetric, # Does the output answer the question? FaithfulnessMetric, # Is the output grounded in the retrieval context? ContextualRelevancyMetric,# Is the retrieved context relevant to the question? HallucinationMetric, # Does the output contain fabricated claims not in the context? GEval, # Custom LLM-judged metric using criteria you define. The criteria is defined via a separate prompt.) For our use case, we will use the Faithfulness metric to check if the actual LLM output is grounded in the retrieval context. from deepeval.metrics import FaithfulnessMetricfaithfulness = FaithfulnessMetric(threshold=0.8) Because the evaluation itself is performed by an LLM, scores may slightly vary between runs. Metric 2: LaBSE semantic similarity (deterministic) from deepeval.metrics import BaseMetricfrom deepeval.test_case import LLMTestCasefrom sentence_transformers import SentenceTransformer, utilclass LaBSESimilarityMetric(BaseMetric): """Deterministic semantic similarity using LaBSE embeddings.""" def __init__(self, threshold: float = 0.7): self.threshold = threshold self.score = 0 self.success = False self.model = SentenceTransformer("sentence-transformers/LaBSE") async def a_measure(self, test_case: LLMTestCase, *args, **kwargs) -> float: return self.measure(test_case) def measure(self, test_case: LLMTestCase) -> float: embeddings = self.model.encode( [test_case.actual_output, test_case.expected_output], convert_to_tensor=True, ) self.score = util.cos_sim(embeddings[0], embeddings[1]).item() self.success = self.score >= self.threshold return self.score def is_successful(self) -> bool: return self.success @property def __name__(self): return "LaBSE Semantic Similarity" The class extends BaseMetric from DeepEval and implements: measure — encodes both actual_output and expected_output into embeddings, computes their cosine similarity, and tests it against the threshold. a_measure — async wrapper required by DeepEval's interface is_successful — returns whether the measurement passed the threshold. For the same inputs, LaBSE always produces the same embeddings, which follow with the same cosine similarity score. As a result, this metric is fully deterministic and reproducible. Running both metrics on one test case from deepeval import assert_testdef test_varnish_response(): assert_test(test_case, [ FaithfulnessMetric(threshold=0.8), # non-deterministic LaBSESimilarityMetric(threshold=0.7), # deterministic ]) Assuming the file containing the mentioned code is named test_varnish.py, the tests can be invoked through the following command: deepeval test run test_varnish.py Running the command produces the output: The test passes only if both metrics meet their thresholds. The faithfulness check ensures the output doesn't hallucinate beyond the context, while the LaBSE similarity check ensures it remains semantically close to the expected answer. Conclusion LLM-as-a-Judge can be used for free-form text where it is hard (or sometimes impossible) to define the correctness of an answer. (summaries, explanations, faithfulness, etc.) Deterministic metrics should be used for tasks with clearly defined correct outputs and where reproducible scores are needed. Another advantage of deterministic metrics is the avoidance of LLM-judge-related costs. In practice, the most effective evaluation suites combine both approaches. DeepEval provides a bridge between the familiar world of Pytest and the modern requirements of LLM testing. By combining LLM-judged metrics for subjective quality assessment with deterministic metrics for measurable correctness, it provides a complete evaluation toolkit that can be easily run from the CLI or a CI pipeline. References: Feng, F., Yang, Y., Cer, D., Arivazhagan, N., & Wang, W. (2020, July 3). Language-agnostic BERT sentence embedding. arXiv.org. https://arxiv.org/abs/2007.01852 DeepEval by Confident AI. (2026, April 5). Quick Introduction. https://deepeval.com/docs/getting-started LLM-as-a-Judge Metrics | Confident AI Docs. (n.d.). Confident AI Docs. https://www.confident-ai.com/docs/llm-evaluation/core-concepts/llm-as-a-judge Ip, J. (2025, October 10). LLM-as-a-Judge simply explained: The complete guide to run LLM evals at scale. Confident AI. https://www.confident-ai.com/blog/why-llm-as-a-judge-is-the-best-llm-evaluation-method DeepEval by Confident AI. (2026a, April 5). “Do it yourself” Metrics. https://deepeval.com/docs/metrics-custom Zheng, L., Chiang, W., Sheng, Y., Zhuang, S., Wu, Z., Zhuang, Y., Lin, Z., Li, Z., Li, D., Xing, E. P., Zhang, H., Gonzalez, J. E., & Stoica, I. (2023, June 9). Judging LLM-as-a-Judge with MT-Bench and Chatbot Arena. arXiv.org. https://arxiv.org/abs/2306.05685

Introduction Kubernetes is not just a container orchestration platform. It is a platform that can be extended with custom APIs. In this blog, the following topics will be covered: what the Kubernetes API is and how it can be extended using a Custom Resource Definition (CRD for short), what a Kubernetes operator is and how it can be implemented, a practical example of extending the Kubernetes API using a CRD, a practical example of implementing an operator that manages objects of the new CRD. As part of the hands-on example, a Website CRD and an operator will be implemented. Based on Website objects, the operator will automatically create a Deployment, Service, and Ingress. Kubernetes API Kubernetes is an API-driven platform, meaning all resources in the cluster are managed through the Kubernetes API server. Creating Kubernetes resources (e.g., Deployment, Service, Ingress, etc.) is done by sending a request to the Kubernetes API server, which: validates the object stores it in etcd and makes it available to other Kubernetes components It is important to understand that the API server does not execute changes directly. It only stores the desired state, while background controllers reconcile and try to align the system's current state with it. This model makes Kubernetes a declarative system so the user defines what they want, and Kubernetes tries to achieve it. How CRDs Extend the Kubernetes API Although Kubernetes comes with a large number of predefined resources, there is often a need to create custom types of objects, like: Database KafkaCluster BackupPolicy Website A CRD allows you to create new types of resources without modifying the Kubernetes source code. After creating a CRD, the new resource becomes part of the Kubernetes API and can be used through: kubectl YAML manifests API calls just like standard Kubernetes resources. CRDs therefore, turn Kubernetes into an extensible platform capable of managing resources from any domain. It is important to understand the difference between a CRD, which defines only the schema of a new Kubernetes resource, and a Custom Resource (CR for short), which represents an instance (an object) of that CRD. Kubernetes Operator and reconciliation loop A Kubernetes operator is a controller that, by observing a CRD and its instances, implements defined logic and automatically brings the system into the desired state. Its main responsibility is the reconciliation loop, which continuously compares the desired state defined in the custom resource with the actual state of the system. If there is a difference between the desired and actual state, the operator performs the necessary actions. In the hands-on example, it will be shown how the operator creates the required objects based on a Website resource, and it is going to be: Deployment Service Ingress In addition to being event-driven, meaning it reacts to events related to the resources it watches, an operator is also state-based, meaning that every time it reacts, it tries to bring the system into the desired state. Lifecycle from CRD to Managed Resource The lifecycle of an operator can be described through the following steps: CRD registration The Kubernetes API is extended with a new resource type. Creating a custom resource The user defines the desired state through a YAML manifest. Validation and storage The API server validates the resource based on the CRD schema and stores it in etcd. Change detection The operator receives an event through the watch mechanism. Reconciliation loop The operator compares the desired state with the current state of the system. Creating or updating resources Required resources are created or updated, and they can be inside or outside the cluster. Status update The operator updates the observed state in the CR status section. This process runs continuously throughout the entire lifecycle of the resource. OwnerReferences and Finalizers Operators not only manage resource creation, but also their lifecycle. OwnerReferences OwnerReferences define the relationship between the parent CR (e.g., Website) and child (Deployment, Service, Ingress …) resources. If an operator creates a Deployment, Service, or Ingress and links them to a Website resource, the Kubernetes garbage collector will automatically delete the child resources when the parent is deleted (cascading deletion). However, if a child resource is deleted first, the operator will recreate it during the reconciliation loop, as it always aims to restore the systemto the desired state defined in the Website resource. How it looks in the child resource definition: ownerReferences: - apiVersion: website.demo.atlantbh.com/v1 blockOwnerDeletion: true controller: true kind: Website name: abh-demo-website uid: 9be6fcb1-cf19-40ce-bfed-8f7c13c23057 Finalizers Finalizers allow additional cleanup logic before a resource is permanently deleted. When a resource contains a finalizer: Kubernetes sets the deletionTimestamp the resource remains in the cluster the operator performs cleanup logic Typical examples include: deleting cloud resources DNS cleanup backing up data before deletion Only after the cleanup is complete does the operator remove the finalizer, and Kubernetes permanently deletes the resource. A finalizer can be seen as a blocker during deletion, meaning the resource cannot be deleted until all finalizers are removed, i.e., until the entire cleanup process is finished.nalizer can be seen as a blocker during deletion, meaning the resource cannot be deleted until all finalizers are removed, i.e., until the entire cleanup process is finished. Implementation Using Kubebuilder For implementing a Kubernetes operator, one framework to consider  is Kubebuilder, which is built on top of the controller-runtime library. Kubebuilder provides: generation of CRD and RBAC manifest files controller skeleton controller-runtime integration standardized project structure for operators This keeps the focus on business logic rather than boilerplate code.  The complete source code of the hands-on example of the blog is available in the GitHub repository. Hands-on Example: Website CRD A simple example demonstrates the implementation of a Kubernetes operator and CRD. The complete CRD is available at the following location.Based on the previously defined CRD, the user defines a custom resource through a YAML file: apiVersion: website.demo.atlantbh.com/v1kind: Websitemetadata: name: abh-demo-websitespec: image: nginx:latest replicas: 3 port: 80 env: - name: ENV value: "production" resources: requests: cpu: "100m" memory: "128Mi" limits: cpu: "500m" memory: "256Mi" What Does the Operator Do? After this custom resource is created, the operator runs a reconcile loop: creates a Deployment (image, replicas, env, resources) creates a Service (exposes the port) creates an Ingress (exposing the application) Then, through the reconciliation loop, it ensures: the number of pods matches replicas the resources are available routing is functioning correctly CRD Validation CRDs use OpenAPI v3 schema validation. In this example: image is a required field - repository replicas must be between 0 and 100 - repository port must be between 1 and 65535 - repository In this way, invalid resources are rejected by the Kubernetes API before they ever reach the operator. Conclusion Custom Resource Definitions enable extending the Kubernetes API and creating custom resource types within a cluster.  When combined with the operator pattern, Kubernetes becomes a platform for automating complex systems, not just container orchestration. Kubebuilder significantly simplifies operator development, allowing developers to focus on business logic rather than infrastructure boilerplate code. This example demonstrates how a simple Website resource can become a declarative interface for automatically managing a web application.

A story about adding a new column to a table with 600 million records The Challenge Picture this. You have a task to implement data loading from a large PostgreSQL database table to Snowflake through a data pipeline. Simple enough, right? Just do it incrementally, using the appropriate column as a bookmark to track how much data has already been processed. The problem is that the table has 600 million records in production and lacks a column suitable for the bookmark. This was exactly the situation we faced recently. We needed to: Add an updated_at column to track changes. Backfill existing records with updated_at = created_at Add an index on the updated_at column for efficient querying (essential for bookmark-based incremental loads) Do all of this without affecting production traffic. Why Traditional Approaches Don't Work Even though created_at was there, it didn’t fit. The updated_at timestamp was the most suitable choice for the bookmark because a single record can be updated over time, and we need to capture those updates.  The naive approach would be to create a new migration with an ALTER TABLE statement, followed by an UPDATE to populate the column, and then CREATE INDEX: ALTER TABLE tableADD COLUMN updated_at TIMESTAMP DEFAULT now();UPDATE tableSET updated_at = created_at;CREATE INDEX idx_table_updated_atON table(updated_at); The whole migration would be wrapped inside a transaction by default.  But what happens next? Adding a column with a default value updates each existing row. This can take hours and locks all rows until it’s done. An UPDATE without batching creates a large transaction that locks all rows for an extended period. Even without the transaction, the lock stays until the statement finishes. This can take a while. So, moving the statement to a separate migration won’t help much. A CREATE INDEX operation gets an exclusive lock on the whole table. This stops all writes and reads for hours.  Any of these operations could bring production to its knees. The table size is the real issue. With less data, all these operations would finish quickly and not disrupt the system. Locking becomes a problem when the operation takes too long. The Zero-Downtime Strategy Here’s how we addressed the issue using PostgreSQL's features and strategic batching. Step 1: Add the Column without a default value We began by adding the column without immediately populating it. In PostgreSQL, adding a nullable column without a default value is almost instantaneous: ALTER TABLE table ADD COLUMN updated_at TIMESTAMP; Step 2: Batch Update Existing Records Rather than updating all 600 million records in a single transaction, we processed them in smaller chunks of 10,000 records. This method: Keeps transactions small and efficient Allows other operations to continue between batches Prevents lock escalation DO 'DECLARE chunk_size INT := 10000; start_id INT; max_id INT;BEGIN SELECT MIN(id), MAX(id) INTO start_id, max_id FROM table; WHILE start_id <= max_id LOOP EXECUTE format( ''UPDATE table SET updated_at = created_at WHERE id >= %s AND id < %s'', start_id, start_id + chunk_size ); start_id := start_id + chunk_size; END LOOP;END'; Step 3: Create Index Concurrently PostgreSQL's CREATE INDEX CONCURRENTLY is a valuable option in these scenarios.  While index creation is slower than usual, it offers greater safety for production environments. Unlike regular index creation, it: Does not block reads and writes Performs multiple table scans to build the index Takes longer but ensures full availability CREATE INDEX CONCURRENTLY idx_table_updated_at ON table (updated_at); Important note: This command must be executed outside of a transaction block. Running it within a transaction would negate its benefits and cause blocking. The complete migration script now looks like this: ALTER TABLE table ADD COLUMN updated_at TIMESTAMP;DO 'DECLAREchunk_size INT := 10000;start_id INT;max_id INT;BEGINSELECT MIN(id), MAX(id) INTO start_id, max_id FROM table;WHILE start_id <= max_id LOOPEXECUTE format(''UPDATE table SET updated_at = created_at WHERE id >= %s AND id < %s'',start_id, start_id + chunk_size);start_id := start_id + chunk_size;END LOOP;END';CREATE INDEX CONCURRENTLY idx_table_updated_at ON table (updated_at); Implementation with Liquibase As our backend is built on Spring Boot, we utilize Liquibase for database migrations. Below is a complete migration file that reflects what we implemented in the production environment: <?xml version="1.1" encoding="UTF-8" standalone="no"?><databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.5.xsd"> <changeSet id="add-table-updated_at" author="amahovac" context="post-release"> <addColumn tableName="table"> <column name="updated_at" type="TIMESTAMP WITH TIME ZONE"/> </addColumn> </changeSet> <changeSet id="update-table-updated_at" author="amahovac" runInTransaction="false" context="post-release"> <!-- Batch update logic here --> </changeSet> <changeSet id="create-concurrent-index-table-updated_at" runInTransaction="false" author="amahovac" context="post-release"> <!-- Concurrent index creation here --> </changeSet></databaseChangeLog> Key Details Separate migrations We organized operations into distinct migrations (changesets) to improve control over each. Running migrations outside the transaction This is crucial for concurrent operations. If not managed, Liquibase would encompass everything in a transaction, leading to the locks we aim to avoid. Adding runInTransaction="false" prevents this issue. Skipping migrations during deployment We used context="post-release" and adjusted the Spring Boot configuration files to set spring.liquibase.contexts=!post-release before deployment.  This way, migrations won't run during deployment and potentially time out our CodeDeploy process. Running it manually After deployment, we executed these migrations manually during a low-traffic period to monitor performance impacts and minimize risk. The Result The entire operation was completed successfully with zero downtime. The production application continued serving requests throughout the 5+ hour process. Our data pipeline now efficiently syncs the table incrementally to Snowflake.  Lessons Learned Test with production-sized data: Our initial approach performed well in the development environment with 1 million rows, but required significant adjustments for the production database. Always consider data volume and test with realistic datasets. Monitor during execution: We kept an eye on CPU, I/O, and lock metrics throughout the process. Document everything: Future team members need to understand the nuances of these migrations. Our inline comments saved confusion later. Always include created_at and updated_at columns when creating a table: You never know when they might be needed. If we had included them from the start, we could have avoided all these complications. Avoid unnecessary locking: Consider concurrent solutions and batch large updates to minimize lock escalation. Run DDL operations with caution: If possible, execute long-running migrations outside of transactions and separate them from deployment.  Working with large production database tables requires patience, careful planning, and an awareness of the impact of each operation. The extra effort to ensure zero downtime is always worthwhile when users don't even notice you just modified 600 million records.