In today's data-driven landscape, ensuring the reliability and accuracy of your data warehouse is paramount. The cost of not testing your data can be astronomical, leading to critical business decisions based on faulty data and eroding trust. 

The path to rigorous data testing comes with its own set of challenges. In this article, I will highlight how you can confidently deploy your data pipelines by leveraging Starlake JSQLTranspiler and DuckDB, while also reducing costs. we will go beyond testing your transform usually written in SQL and see how we can also test our Ingestion jobs.

The art of mastering data pipelines​

Mastering your data pipeline is a challenging art. A data pipeline generally contains the following phases:

• Collection: Extracting data from sources
• Ingestion: Loading the extracted data into the data warehouse
• Transformation: A phase that ultimately adds value to the collected data

The table below summarizes the tests run by Starlake on Load & Transform jobs:

Untested SQL costs​

Thanks to data pipelines unit testing, we drastically reduce the development cost. Not running tests seems to allow high project's velocity, thereby delivering value quickly. However, the cost of a feature does not stop at its simple development but encompasses all the efforts put in until the feature's completion. Below are some hidden costs.

• Identifying bugs
  • Without unit tests, verifying development requires deploying the project. This deployment raises challenging questions about the deployment strategy and its rollback or correction procedures.
  • Verification might be carried out by a separate QA team, sometimes even outside the project team. This can lead to the use of feature flags to avoid deploying to production, complicating the implementation. Additionally, waiting for feedback from the QA team introduces delays, increasing the cost of fixing any bugs that arise.
  • Depending on the deployment strategy, verification may also be incomplete due to a lack of control over the test data used.
• Maintenance and Evolution Complexity
  • Many of us have faced a massive query and struggled to make modifications without disrupting existing functionality, all while aiming for improvements like optimizing processing time. Rigorous unit tests can help with this. They allow us to enhance the expected outcomes in current datasets, create new ones, and compare the modified query results with these expectations. This significantly reduces the risk of regression.
• Decreased productivity
  • The absence of automated tests often means manually re-running parts or all of the system to ensure correct integration, which can lead to spending time fixing collateral bugs and thus reducing overall productivity. As the project advances, more components need verification, making the process even more time-consuming. This significantly diminishes the willingness to refactor or revise code.
• Promoting expertise
  • Without unit tests, teams often assign the same tasks to the same people, which hinders skill development and increases the risk of knowledge loss due to turnover.
• Customer dissatisfaction
  • A project with uncertain product output quality often leads to dissatisfaction, frustration, and a loss of trust in the individual, the team, or the product.

We are all aware of the hidden costs associated with the absence of tests; in my opinion, these are the most significant. Therefore, we will explore how to manage a data pipeline.

Writing unit test in Starlake​

Suppose we have the following transform.

We test it by creating the following hierarchy:

This is then how it is executed

Starlake unit tests benefits​

Running tests on a local DuckDB database instead of the target Data Warehouse has the following advantages:

• Fast Feedback: Local execution is significantly faster than using a remote database due to network latency. Additionally, the local environment might be better suited for handling small volumes of test data.
• No Execution Cost: Depending on the pricing model of the target database, creating temporary resources and executing queries can incur both execution and storage costs.
• Setup and Cleanup of Automated Tests: Guarantee of resource isolation.
• Credential Issues: Running tests against a target database requires credentials, which may pose security risks.

Conclusion​

In this article, we have demonstrated how adopting unit testing, a crucial practice for software engineers, can significantly enhance the quality of our data pipelines. This approach not only reduces overall costs in the medium to long term but also ensures the maintenance of dynamic and enduring documentation. Additionally, implementing unit tests is essential for rigorous CI/CD processes, enabling seamless continuous data pipeline deployment.

If you encounter any issues while performing your tests locally, please report them on the Starlake GitHub repository. Your feedback is invaluable in improving local test coverage, empowering more data engineers to deploy their work confidently and smoothly. For further discussions and support, join our team on Slack.

We greatly appreciate your contributions. If you found this article helpful, please star the project on GitHub and share it on your social networks to help us reach a broader audience.

Data exposition strategies​

Data may be exposed using views or authorized views and more recently using Row / Column level security.

Historically, to restrict access on specific columns or rows in BigQuery, one can create a (authorized) view with a SQL request like the one below:

BigQuery Views require to grant access for the end users to the table on top of which the view is created. To bypass that limitation, BigQuery provide Authorized views. However, Authorized views come with the following restrictions:

1. The underlying table is accessed through the authorized view where the end user is impersonated, loosing thus at the table level, the identity of the user making the request.

2. Each restriction policy require to define a specific authorized view making it difficult to identify who has access to what ?

3. Authorized views need to be updated whenever a schema evolution on the underlying table bring in a sensitive field that need to be excluded or a field that need to be included in the view. In the example below, the new column "description" need to be added to the authorized view if we want it .

That's where Row Level Security and Column Level security features natively supported by BigQuery come in.

BigQuery Row Level Security​

Row Level Security restrict access to the rows based on the conditions set in the where clause using the custom SQL statement below:

Big Query Column Level Security​

Column level security in BigQuery is managed using a taxonomy. This taxonomy is a hierarchy of policy tags describing the table attributes or other resources. By assigning access rights to a tag, we restrict access to any resource tagged using this specific tag and this applies to BigQuery table fields.

In our example, restricting access to specific user/group/sa to the column price require the following steps:

1. In Cloud Data Catalog/Policy Tags, create a Taxonomy. Note that Enfore access control should be checked.

2. Assign permissions for each policy tag you defined

3. Tag restricted columns in the BigQuery schema editor.

BigQuery RLS/CLS benefits​

Using BigQuery row and column level security features bring several benefits:

• There is no need to create extra views
• Users use the same name for the table but with different access rights
• A company-wide taxonomy is defined allowing better Data Management
• Access rights to a new column in the table are automatically handled

A word about RLS and CLS in Starlake​

Ingesting Data into BigQuery cannot be considered complete without taking into account the access level restrictions on the target table. Starlake will handle for you all the scripting required to secure BigQuery rows and columns using a YAML declarative syntax to make sure that your tables are secured in BigQuery:

  - name: "PRODUCT"
    rls:
      - name: "my-rls"
        predicate: "category like 'Food'"
        grants:
          - "user:me@company.com"
          - "group:financegroup@company.com"
          - "sa:serviceacount@gserviceaccount.com"
    attributes:
      - name: "id"
        accessPolicy: PII

Data Loading strategies​

When loading data into BigQuery, you may want to:

• Overwrite the existing data and replace it with the incoming data.
• Append incoming data to existing
• Dynamic partition Overwrite where only the partitions to which the incoming data belong to are overwritten.
• Merge incoming data with existing data by keeping the newest version of each record.

For performance reasons, when having huge amount of data, tables are usually split into multiple partitions. BigQuery supports range partitioning which are uncommon and date/time partitioning which is the most widely used type of partitioning. The diagram below shows our initial table partitioned by the date field.

Let's assume we receive the following data that we need to ingest into the table:

The strategies above will produce respectively the results below:

There is no good or bad strategy, the use of one of the strategies above depends on the use case. Some use case examples for each of the strategies are:

• Overwrite mode may be useful when you receive every day the list of all product names.
• Append mode may be useful when you receive daily sales.
• Dynamic Partition Overwrite mode may be useful when you ingested the first time a partition, and you need to ingest it again with a different set of data and thus alter only that partition.
• Merge mode may be useful when you receive product updates every day and that you need to keep only the last version of each product.

Spark How-to

Apache Spark SQL connector for Google BigQuery makes BigQuery a first class citizen as a source and sink for Spark jobs.

Append and Overwrite modes in Spark​

BigQuery is supported by Spark as a source and sink through the Spark BigQuery connector

Spark comes out of the box with the ability to append or overwrite existing data using a predefined save mode:

val incomingDF = ... // Incoming data loaded with the correct schema
val bqTable = "project-id.dataset.table"
val saveMode = SaveMode.Overwrite // or SaveMode.Append fot he appending data
incomingDF.write
    .mode(saveMode)
    .partitionBy("date")
    .format("com.google.cloud.spark.bigquery")
    .option("table", bqTable)
    .save()

Dynamic Partition Overwrite mode in Spark​

To activate dynamic partitioning, you need to set the configuration below before saving the data using the exact same code above:

spark.conf.set("spark.sql.sources.partitionOverwriteMode","DYNAMIC")

Unfortunately, the BigQuery Spark connector does not support this feature (at the time of writing). We need to manually delete the partitions we want to overwrite first and then append the incoming data.

Assuming the table is partitioned by the field date and the incoming data loaded in the incomingDF dataframe, the code below will remove existing partitions that need to be overwritten.

val incomingDF = ... // Incoming data loaded with the correct schema
incomingDF
  .select(date_format(col("date"), "yyyyMMdd").cast("string"))
  .distinct()
  .collect()
  .map(_.getString(0))
  .foreach { partition =>
    bigQueryClient.deleteTable(TableId.of(datasetName, s"$table\$$partition"));
  }

bq rm -t 'project-id.dataset.table$YYYYMMDD'

We now need to append the incomingDF to mimic the dynamic partition overwrite feature:

val incomingDF = ... // Incoming data loaded with the correct schema
val bqTable = "project-id.dataset.table"
val saveMode = SaveMode.Append
incomingDF.write
    .mode(saveMode)
    .partitionBy("date")
    .format("com.google.cloud.spark.bigquery")
    .option("table", bqTable)
    .save()

Dynamic Partition Merge in Spark​

When you need to keep the last version of the record for each product, both BigQuery and Databricks (the company behind Spark in case you lived on the moon the last ten years) support the merge SQL statement:

MERGE INTO target_table
USING incoming_table
ON target_table.product = incoming_table.product
    WHEN NOT MATCHED
		THEN INSERT *
	WHEN MATCHED AND incoming_table.date > target_table.date THEN
		UPDATE SET *
/*
    WHEN MATCHED AND incoming_table.timestamp <= target_table.timestamp THEN
		SKIP
*/

Unfortunately the MERGE statement is not supported by Apache Spark. It is only supported by Databricks, its commercial version.

To do a merge using the Spark BigQuery connector, we need to do it by following the steps below:

Step 1: Create a dataframe with all the rows

val allRowsDF =
    incomingDF
    .unionByName(existingDF)

Step 2: group by product and order each product occurrence by date descending

val orderingWindow =
    Window
      .partitionBy("product")
      .orderBy(col("date").desc, col("product")))

val orderedDF =
    allRowsDF
        .withColumn("rownum", row_number.over(orderingWindow))

In the step 2 above, each product is ordered by date with the most recent one first (descending order). We identify it by the rownum column.

Step 3: Keep the most recent product

val toKeepDF =
    orderedDF
        .where(col("rownum") === 1)
        .drop("rownum")

Step 4: Overwrite existing partitions with the data we want to keep

val bqTable = "project-id.dataset.table"
val saveMode = SaveMode.Overwrite
toKeepDF.write
    .mode(saveMode)
    .partitionBy("date")
    .format("com.google.cloud.spark.bigquery")
    .option("table", bqTable)
    .save()

Starlake How-to​

Starlake is a declarative Ingestion Framework based on YAML description files.
The 4 ingestion strategies described above are supported through the settings below:

     name: "mydb"
     directory: "..."
+    metadata:
    schemas:
      - name: "mytable"
        pattern: "data-.*.csv"
        metadata:
          writeStrategy:
            type: "OVERWRITE"
       attributes:
          - name: "date"
            type: "date"
            rename: "id"
          - name: "product"
            type: "string"
          - name: "price"
            type: "decimal"

     name: "mydb"
     directory: "..."
+    metadata:
    schemas:
      - name: "mytable"
        pattern: "data-.*.csv"
        metadata:
          writeStrategy:
            type: "APPEND"
        attributes:
          - name: "date"
            type: "date"
            rename: "id"
          - name: "product"
            type: "string"
          - name: "price"
            type: "decimal"

    name: "mydb"
    directory: "..."
+   metadata:
    schemas:
      - name: "mytable"
        pattern: "data-.*.csv"
        metadata:
          strategy:
            type: "OVERWRITE_BY_PARTITION"
          sink:
            partition:
              - "date"
       attributes:
          - name: "date"
            type: "date"
            rename: "id"
          - name: "product"
            type: "string"
          - name: "price"
            type: "decimal"

     name: "mydb"
     directory: "..."
+    metadata:
    schemas:
      - name: "mytable"
        pattern: "data-.*.csv"
        metadata:
          strategy:
            type: "OVERWRITE_BY_KEY_AND_TIMESTAMP"
            key: ["product"]
            timestamp: "date"
        attributes:
          - name: "date"
            type: "date"
          - name: "product"
            type: "string"
          - name: "price"
            type: "decimal"