GCP Data Engineering Project: ELT Financial Data Pipeline with dbt, Cloud Run, Composer/Airflow, BigQuery, and Dataplex 🪙

In this project, I'll walk you through building an ELT financial data pipeline using Google Cloud services, with a special focus on three powerful tools: dbt, Apache Airflow, and Dataplex.

Why these tools? If you browse data engineering jobs on LinkedIn, you'll notice that about 80% mention Airflow and dbt. So why not practice both together? In this project, I'll show you the most efficient way to run dbt on GCP while Composer/Airflow orchestrates the entire pipeline.

There's another important reason for this project. After joining the financial sector, my perspective on data quality completely changed. It went from being important to absolutely critical - in finance, it's non-negotiable. That's where Dataplex becomes essential.

This project will show you how to integrate Dataplex into your pipeline to catch quality issues early, automate validation checks, and build confidence in your data.

---

Services I used in this project:

🗃️ GCS bucket: Stores raw data files for the pipeline

⚡Cloud Run function: Monitors GCS for new file uploads and automatically triggers the Airflow DAG

✨ Composer/Airflow: Orchestrates the entire pipeline, managing task dependencies and execution flow

🧮 BigQuery: Central data warehouse that stores both raw and transformed data for analysis

🔄 Cloud Run/dbt: Executes containerized dbt jobs for data validation and transformations (two test jobs + one transformation job)

🏗️ Cloud Build: Automates building and deploying Docker containers for dbt Cloud Run jobs

📦 Artifact Registry: Stores versioned Docker images containing dbt models and transformation logic

🔍 Dataplex: Performs automated data profiling, quality scans, and validation checks on transformed datasets

📧 Gmail: Sends email notifications for pipeline status updates and data quality scan results

ELT Financial Data Pipeline

Below, you can see how the entire pipeline works within the DAG. This is what we'll build step-by-step.

This project demonstrates an automated data pipeline that processes incoming data files without requiring predefined schedules. When data arrives in Cloud Storage, whether from external systems, APIs, or manual uploads, a Cloud Run function detects the new file and triggers an Airflow DAG to begin processing.

The main DAG elt_financial_data_pipeline orchestrates several tasks: it loads raw data into BigQuery, archives the processed file to a separate GCS folder, and executes three Cloud Run jobs. These containerized jobs perform data validation and transformation using dbt.

Upon successful completion, the pipeline sends a notification email and automatically triggers a secondary DAG dataplex_etl_with_quality_checks_and_profile_scan that runs Dataplex profile and data quality scans. Results from these scans are also delivered via email, providing immediate visibility into data quality metrics.

---

Note that the data (synthetic_financial_data_20251224.csv) presented is synthetically generated by AI and does not represent real information. It was created to include more columns and values for experimentation during transformations. 

Columns:

        ID1, ID2, ID3 - alphanumeric IDs (distinct ID spaces with prefixes A, B, C)
        - String1 - business category (Retail, Wholesale, Online, Subscription, B2B, Marketplace)
        - String2 - currency-like codes (EUR, USD, GBP, PLN, SEK, NOK) with a heavier share of EUR
        - String3 - reference-style token (REF-xxxxxxxx)
        - Float1 - "amount" (skewed positive values, mostly low to mid with a long tail)
        - Float2 - fee (roughly 0–3% of amount + noise, non-negative)
        - Float3 - net (≈ amount − fee, clipped to remain ≤ amount and ≥ 0)
        - Status - categorical status (Settled, Pending, Failed, Reversed, Cancelled, In Review) with realistic proportions

I'll be honest, I got a little overconfident with the AI. I assumed the data would be flawless and jumped straight into building without checking it first. When Dataplex flagged quality issues at the end, it was a good wake-up call. 😅 Lesson learned: always validate your data, no matter where it comes from. Quality checks aren't optional. The more checks you have in place, the better!

dbt with Cloud Run ♾️

There are several ways to run dbt on Google Cloud. You can install dbt locally (pip install dbt-bigquery) and connect directly to BigQuery with minimal cost and full features. However, this isn't production-ready: it requires manual execution, lacks centralized logging and automation, poses security risks with local credentials, and creates version inconsistencies across teams. It's fine for development and learning, but not for production.

The better option for production is packaging dbt in Docker containers and deploying as Cloud Run serverless jobs. This gives you container isolation, version-controlled environments, pay-per-execution pricing (no idle costs), fast startup, seamless CI/CD integration with Cloud Build, and easy rollbacks. Perfect for production workloads, large projects, multiple environments, and teams that need cost efficiency and scalability. However, be aware of Cloud Run's limitations by checking the official docs before committing to this approach.

Start by preparing your dbt files, Dockerfile, cloudbuild.yaml, and other necessary files. The complete code is available here.

Here's an example of a containerized dbt project structure:

dbt-project/
├── dbt_project.yml          # Required: Project configuration and name
├── profiles.yml             # Required: Database connection details
├── packages.yml             # External dbt packages (optional)
|
├── models/                  # Required: SQL files, tests, and model descriptions
│   ├── model.sql            # Transformation logic
│   └── schema.yml           # Contains declarative tests and descriptions for models.
|
├── macros/                  # Custom Jinja macros (optional)
├── tests/                   # Custom data tests (optional)
├── seeds/                   # CSV files to load as tables (optional)
├── snapshots/               # SCD Type 2 tracking (optional)
├── analyses/                # Ad-hoc queries and reports (optional)
├── docs/                    # Generated documentation (optional)
|
├── .dockerignore            # Files to ignore in Docker build
├── Dockerfile               # Docker image build instructions
├── cloudbuild.yml           # CI/CD pipeline steps
└── requirements.txt         # Python dependencies

Here are my 3 Cloud Run jobs:

dbt-test-raw-job: Runs tests on raw data that has been loaded into BigQuery. Deploy this job: gcloud builds submit --config cloudbuild.test.raw.yml .
dbt-transform-job: Executes transformations on raw data in BigQuery. Deploy this job: gcloud builds submit --config cloudbuild.transform.yml .
dbt-test-transformed-job: Conducts tests on the transformed data in BigQuery. Deploy this job: gcloud builds submit --config cloudbuild.test.transformed.yml .

These commands build your image in Cloud Build, pushes it to Artifact Registry, and deploys it to Cloud Run as a job.

Here's how the Docker images appear in Artifact Registry:

And here's my Cloud Build history. Each build gets a unique ID:

I divided my dbt workflow into three Cloud Run jobs to run them as separate Airflow tasks.

This separation ensures the correct order: raw data validation → transformation → transformed data validation. Each step has clear success or failure states, making it easy to identify which specific stage failed in the Airflow UI. While a single job is technically possible, this approach provides better observability, simplifies debugging, and gives you fine-grained control over the pipeline flow.

Click on any job to view details. Click "Execute" to run it manually, or click "View" to open the logs and see all processes and results.

The logs show all processes, results, and whether tests passed or failed.

Note: Cloud Run can connect directly to GitHub, GitLab, or Bitbucket repositories for CI/CD automation. Check the Google Cloud docs for setup details.

Here is another example, in Cloud Run job dbt-test-transformed-job which tests the transformed data, there are tests with severinity error/warn - what is the difference? I have 2 tests for fee_percentage.

Column-level test:

- dbt_expectations.expect_column_values_to_be_between:
    arguments:
      min_value: 0
      max_value: 5

Table-level test:

- dbt_utils.expression_is_true:
    arguments:
      expression: "fee_percentage <= 5.0 OR gross_amount = 0"

Those tests give warn in logs.

And if you decide to check your transformed data in BigQuery:

-- Overall data quality summary
SELECT 
  COUNT(*) AS total_rows,
  SUM(CASE WHEN merchant_id IS NULL THEN 1 ELSE 0 END) AS null_merchant_id,
  SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) AS null_customer_id,
  SUM(CASE WHEN final_status IS NULL THEN 1 ELSE 0 END) AS null_final_status,
  SUM(CASE WHEN fee_amount > gross_amount THEN 1 ELSE 0 END) AS fee_exceeds_gross,
  SUM(CASE WHEN fee_percentage > 5.0 THEN 1 ELSE 0 END) AS high_fee_percentage,
  SUM(CASE WHEN calculated_net > gross_amount THEN 1 ELSE 0 END) AS net_exceeds_gross
FROM `your-project-id.financial_data_dev.transformed_data`;

✅ If all rows pass → Pipeline continues normally

⚠️ If some rows have fee_percentage > 5% → Test fails but warns → Pipeline still runs

You see a warning in the logs but the pipeline doesn't stop.

Orchestration with Cloud Composer ✨

The main question is: how do we tie everything together when using Cloud Run jobs? This is where Composer comes to the rescue. Composer is Google Cloud's managed Airflow service, and for this project, I'm using Composer 3 with the CloudRunExecuteJobOperator organized into Task Groups. This approach provides cleaner visual structure and makes it easier to modify the pipeline later - say, if we want to skip certain failed tests or send notifications without blocking the entire workflow.

Here's an example of how it works:

# Task Group 1: Test Raw Data
    with TaskGroup("test_raw_data_group", tooltip="Test raw data in BigQuery") as test_raw_data_group:
        test_raw = CloudRunExecuteJobOperator(
            task_id='execute_test_raw_data_job',
            project_id='your-project-id',
            region='europe-west1',
            job_name='dbt-test-raw-job',
        )

        # This task runs only if test_raw succeeds
        log_test_raw_success = PythonOperator(
            task_id='log_test_raw_success',
            python_callable=log_job_status,
            op_kwargs={
                'status': 'SUCCESS ✅',
                'job_name': 'dbt-test-raw-job'
            },
            trigger_rule='all_success',
        )
        
        # This task runs only if test_raw fails
        log_test_raw_failure = PythonOperator(
            task_id='log_test_raw_failure',
            python_callable=log_job_status,
            op_kwargs={
                'status': 'FAILED ❌',
                'job_name': 'dbt-test-raw-job'
            },
            trigger_rule='one_failed',
        )

        # A dummy task to join the success/failure branches
        join_raw = DummyOperator(
            task_id='join_raw',
            trigger_rule='all_done',
        )

        test_raw >> [log_test_raw_success, log_test_raw_failure] >> join_raw

The complete DAG is available here: dag-elt-pipeline.py

Below you can see the DAG in action. Each TaskGroup includes four tasks with straightforward logic: if the main task (like execute_test_raw_data_job) succeeds, it turns green, the success log task runs, and the failure task gets skipped (shown in pink).

The EmptyOperator doesn't perform any actual work - it acts as a control flow tool that merges execution branches. After the test runs, the pipeline splits into success and failure paths. The join_raw operator brings these branches back together, creating a single exit point. This simplifies dependencies: the next TaskGroup can depend on this one join task instead of managing complex dependencies on both the success and failure branches.

But what if the test fails and you don't actually care? What if you don't want a failed test to block your downstream tasks?

This is where the join_raw trigger rule becomes important. By setting it to all_done, the join task runs regardless of whether upstream tasks succeeded or failed - it only cares that they've finished. This allows the pipeline to continue even when tests fail.

Here's an example: I intentionally failed execute_test_raw_data_job by changing the "Status" value in the raw data CSV to something not defined in the dbt test's accepted_values ('Settled', 'Pending', 'Failed', 'Reversed', 'Cancelled', 'In Review'). The test failed, the failure was logged, but the pipeline didn't stop because join_raw has trigger_rule='all_done'. The transformation and subsequent tasks continued running, and I still received an email confirming successful pipeline completion.

This pattern is useful for non-critical validations where you want to log issues without blocking the entire data pipeline. Note that other join tasks in this DAG use trigger_rule='none_failed' because those validation steps are critical - if they fail, the pipeline should stop. For more on trigger rules and their behavior, check out the Astronomer docs.

Alternative approach: For more explicit branching where you want only one path to execute (with pink "skipped" status for the path not taken), you can use BranchPythonOperator. Learn more about branching in the Astronomer docs.

This is how my pipeline ends. You can adapt it with your own logic, options, and notifications.

By the way, have you ever seen a DAG task turn purple? Neither had I - until now!

This happens because in the elt_financial_data_pipeline DAG, we're using the TriggerDagRunOperator to trigger another DAG - dataplex_etl_with_quality_checks_and_profile_scan. This puts the task into a Deferred state. Deferred means the task has paused its execution, released its worker slot, and submitted a trigger to be picked up by the triggerer process, meaning the task is waiting without consuming unnecessary resources. This is actually a more efficient way to handle long-running tasks! You can read more about the Deferred state in the Astronomer docs.

Email Alerts and Log Notifications 🔔

I also wanted to make logs more user-friendly. Anyone who's new to data engineering knows that parsing through logs can be overwhelming, so here is an example of how clean and formatted notifications can be added at the end of the pipeline.

Both success and failure tasks can have similar notifications, so it's immediately clear what happened.

---

Let's make this DAG even more useful by adding email notifications!

Email notifications play a crucial role in both security and operational efficiency. They provide essential visibility, allowing you to catch issues immediately upon receiving an alert, ensuring you stay informed about your pipeline's functionality.

Note: A personal Gmail account is being used here. Production environments should use enterprise email infrastructure with proper authentication and monitoring.

Step 1: Generate a Google App Password

Go to your Google Account settings.
Navigate to Security.
Under "How you sign in to Google" make sure 2-Step Verification is turned on. You cannot create App Passwords without it.
At the bottom of the 2-Step Verification page, click on App passwords.
When prompted, select "Mail" for the app and "Other (Custom name)" for the device. You can name it something like "Airflow".
Google will generate a 16-character password. Copy this password immediately, as you won't be able to see it again.

Step 2: Configure the Airflow Connection

Now, you need to add this App Password to your Airflow SMTP connection. You can do this through the Airflow UI.
In the Airflow UI, go to Admin -> Connections.
Find the connection with the Conn Id smtp_default and click the edit button (pencil icon). If it doesn't exist, create a new one.
Fill in the connection details as follows:

Conn Id: smtp_default
Conn Type: Email
Host: smtp.gmail.com
Login: Your full Gmail address (e.g., your.email@gmail.com)
Password: The 16-character App Password you generated in Step 1.
Port: 587 (for TLS) or 465 (for SSL). 587 is standard.

4. Click Save.

Step 3: Add SMTP Configuration and Email Backend

Add these as Airflow configuration overrides:

email
    email_backend     airflow.utils.email.send_email_smtp 
smtp
    smtp_host         smtp.gmail.com 
    smtp_starttls     True 
    smtp_ssl          False 
    smtp_port         587 
    smtp_mail_from    <your-email>

Note: For your Production project, you'll need to use Google Cloud Secret Manager to store your passwords. If you'd like more details, I have another project that shows how Secret Manager works. Follow these instructions for Composer 2 or Composer 3, and read more about Email Backend here.

As you can see, I'm using my personal Gmail for both sending and receiving emails:

# Final success task
    send_success_email = EmailOperator(
        task_id='send_success_email',
        to='jana<...>@gmail.com',
        conn_id='smtp_default',
        subject='[SUCCESS] DBT Pipeline: {{ dag.dag_id }}',
        html_content="""
        <h3>DBT Pipeline Completed Successfully</h3>
        <p>DAG: {{ dag.dag_id }}</p>
        <p>Execution Date: {{ ds }}</p>
        <p>Run ID: {{ run_id }}</p>
        <p>Data is ready for downstream consumption.</p>
        """,
        trigger_rule='all_success',
    )

When you authenticate with smtp.gmail.com using your email and an App Password, Gmail's server enforces a security rule: the "From" address on any email you send must match the address of the account you used to log in.

Cloud Run function ⚡

If it's a new service for you, I would recommend creating the Function manually. Go to Cloud Run -> Services -> Write a function. You can find the code here.

There are a few important code changes you need to make: add the DAG's name and file's prefix. In my case, my DAG is elt_financial_data_pipeline, CSV file prefix is synthetic_financial_data_. Function entry point will stay the same: trigger_dag.

To trigger your DAG, the Function needs to know your Airflow web UI, which can be visible on the Composer "Environment configuration". Don't forget to add it to the code too, as web_server_url. The good news is that this code works with both Composer 2 and Composer 3.

web_server_url = "https://composer-airflow-web-ui-dot-europe-west1.composer.googleusercontent.com"

When creating a trigger ("Add trigger"), pay attention to the Event type. Event type google.cloud.storage.object.v1.finalized means that the Cloud Run function will be triggered when a new object is successfully finalized (uploaded or overwritten) in the configured GCS bucket (in my case, it's "elt-dev"). With all pipeline components ready, simply drop the CSV file into the bucket, and the pipeline will start automatically.

In the DAG elt_financial_data_pipeline, there's a task called move_gcs_files_to_proceed that moves processed files to the "proceed" folder, preventing them from being processed repeatedly. To understand how this works on the Function side, check the logs.

It's important to skip the "proceed" folder in the bucket because our event type google.cloud.storage.object.v1.finalized triggers the Cloud Run function for any new object in the GCS bucket. Without this skip, the Function would be triggered when the "proceed" folder is created.

The Eventarc trigger invocations dashboard will give you an opportunity to see all details, errors, and whether your Function can trigger the DAG.

From my experience, this is unfortunately the most problematic service due to permissions and roles. Enable all the suggested roles shown on the yellow background when you create the Function and the trigger - for example:

This trigger needs the role roles/pubsub.publisher granted to service account service-<your-project-number>@gs-project-accounts.iam.gserviceaccount.com to receive events via Cloud Storage.

Consult the official Google Cloud docs for your specific setup. Based on my implementation, here are all the IAM roles I had to grant for the Function to work:

• Eventarc Service Agent (service-@gcp-sa-eventarc.iam.gserviceaccount.com): Eventarc Service Agent, Cloud Run Invoker

• GCP Project Service Account (service-@gs-project-accounts.iam.gserviceaccount.com): Pub/Sub Publisher

• Your Service Account: Composer Worker, Storage Object Viewer, Cloud Run Invoker, Eventarc Event Receiver

• Cloud Pub/Sub Service Account (service-@gcp-sa-pubsub.iam.gserviceaccount.com): Cloud Pub/Sub Service Agent, Service Account Token Creator. 

For more info, see the Google Cloud docs for complete details and command-line setup.

Dataplex 🔍

You can create many things manually in Dataplex, but I decided to try something more interesting and use Airflow Dataplex operators to trigger data quality scans and data profiling from the DAG. But that's not all: the DAG sends data quality scans via email and creates a table in BigQuery with all the details and errors you need to know. Check the tasks from the Airflow UI.

Let's discuss the Airflow operators for Dataplex:

• DataplexCreateOrUpdateDataQualityScanOperator job is to manage the definition of the data quality scan within Dataplex. This operator ensures that a data quality scan with the ID financial-data-quality-scan exists in Dataplex and that its configuration matches exactly what you've defined in your DAG file. If the scan doesn't exist, it creates it. If it exists, but your DAG has different rules, it updates it.

• DataplexRunDataQualityScanOperator only job is to trigger an execution of an already existing data quality scan. Once your rules are defined, this operator tells Dataplex to run the scan now. It can only trigger a scan that has already been defined.

• DataplexGetDataQualityScanResultOperator retrieves the scan results after execution completes. It pulls the pass/fail status and detailed metrics.

The same pattern applies to Dataplex operators when performing profile scans.

Check this example of TaskGroup for a data quality scan. Find complete DAG here.

# Task Group for Data Quality Scan
with TaskGroup(group_id='data_quality_scan_group') as data_quality_scan_group:
    create_or_update_quality_scan = DataplexCreateOrUpdateDataQualityScanOperator(
        task_id='create_or_update_quality_scan',
        project_id=PROJECT_ID,
        region=REGION,
        data_scan_id=DATA_QUALITY_SCAN_ID,
        body={
            'data': {
                'resource': f'//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/{TABLE_ID}'
            },
            'data_quality_spec': {
                'rules': [
                    {
                        'dimension': 'COMPLETENESS',
                        'name': 'transaction-id-not-null',
                        'description': 'transaction_id must not be null',
                        'column': 'transaction_id',
                        'threshold': 1.0,
                        'non_null_expectation': {}
                    },
                    {
                        'dimension': 'UNIQUENESS',
                        'name': 'unique-transaction-id',
                        'description': 'transaction_id must be unique',
                        'column': 'transaction_id',
                        'threshold': 1.0,
                        'uniqueness_expectation': {}
                    },
                    {
                        'dimension': 'COMPLETENESS',
                        'name': 'reference-token-not-null',
                        'description': 'reference_token must not be null',
                        'column': 'reference_token',
                        'threshold': 1.0,
                        'non_null_expectation': {}
                    },
                    {
                        'dimension': 'UNIQUENESS',
                        'name': 'unique-reference-token',
                        'description': 'reference_token must be unique',
                        'column': 'reference_token',
                        'threshold': 1.0,
                        'uniqueness_expectation': {}
                    },
                    {
                        'dimension': 'COMPLETENESS',
                        'name': 'customer-id-not-null',
                        'description': 'customer_id must not be null',
                        'column': 'customer_id',
                        'threshold': 1.0,
                        'non_null_expectation': {}
                    },
                    {
                        'dimension': 'VALIDITY',
                        'name': 'fee-percentage-in-range',
                        'description': 'Fee percentage should be between 0 and 100',
                        'column': 'fee_percentage',
                        'threshold': 1.0,
                        'range_expectation': {
                            'min_value': '0',
                            'max_value': '100'
                        }
                    },
                ],
                'post_scan_actions': {
                    'bigquery_export': {
                        'results_table': f'//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/dq_results'
                    }
                }
            },
            'execution_spec': {
                'trigger': {
                    'on_demand': {}
                }
            }
        }
    )

The detailed data quality scan results are available via Airflow's XCom mechanism. To receive email alerts with these results, you can use the EmailOperator. The Composer backend and operator logic discussed earlier apply here.

One important thing to understand when using Dataplex operators in Airflow is that the DAG's job is to trigger and run the data quality scan - not to enforce the results. This means that even if Dataplex detects data quality issues like duplicates or missing values, the DAG itself will still show as successful. The scan ran, the results were captured, and Dataplex did its job.

That's why I'm using the EmailOperator to receive scan results via email through XCom, so I can still review them manually and not miss any issues. If you want your DAG to actually fail based on quality results, you need to add additional logic, for example, checking the scan results in a subsequent task and raising an error if the overall score didn't pass.

There are other ways to receive scan results via email. Although configuring everything in code can be challenging, you can simply edit an existing scan through the console and add your email and triggers to "Notification report".

You'll get a detailed email report showing the job overview and which rules failed. The email includes a direct link to open the Dataplex scan.

Additional options include defining data quality rules via gcloud CLI in YAML format or setting up Cloud Logging alerts to notify you of data quality issues. To learn more about Logging and Monitoring, see my previous project or the official docs - and don't hesitate to experiment!

One of the most interesting features is being able to save data quality scan results straight to BigQuery. The DAG handles this automatically with a specific code section.

'post_scan_actions': {
    'bigquery_export': {
        'results_table': f'//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/dq_results'
    }
}

Just a few lines of code, but the results in BigQuery are impressive.

I saved my scan results to a dq_results table, and honestly, it's been a game-changer. You get everything in one place: where the data lives (lake/zone/region), what quality rules you're checking, whether it passed or failed, and the overall score. Instead of hunting through different dashboards, you can query one table and see the full picture of your data health. This allows you to easily query, visualize, and build alerting on your data quality metrics over time.

---

Let's see what was checked. The Airflow DAG ran 6 data quality rules on a financial dataset:

We can see that the column transaction_id failed the uniqueness test.

Let's analyze what actually happened.

⚠️ The Problem: Duplicate Transaction IDs

        Failed Rule: unique-transaction-id
        Pass Ratio: 99.53% (0.9953)
        Failing Records: 47 out of 10,000

❌ UNIQUENESS (50%) - Failed!

        Duplicate detection:
        transaction_id: Has duplicates (47 failing records) ✗
        reference_token: Unique ✓

Why results show 50% of Uniqueness? 1 out of 2 uniqueness checks failed (transaction_id failed, reference_token passed)

🚨 Business Impact

        For financial data, duplicate transaction IDs could mean:
        Double-counting transactions in reports
        Incorrect financial totals (revenue, fees, etc.)
        Reconciliation failures with external systems
        Compliance issues if audited

Priority: HIGH - This should be fixed before the data is used for financial reporting. In reality, you should investigate the root cause and add deduplication logic before loading the data. I couldn't believe my eyes! As I mentioned in the beginning, I expected the data to be perfect. It passed my dbt tests, but Dataplex caught issues I'd missed. I had to investigate and find the duplicate transaction IDs. Then I ran this query in BigQuery:

SELECT
    transaction_id,
    COUNT(transaction_id) AS occurrence_count
FROM
    `your-project-id.financial_data_dev.transformed_data`
GROUP BY
    transaction_id
HAVING
    occurrence_count > 1;

The Dataplex results were correct. The data contained 47 duplicates.

You might be wondering: if dbt tests have already run, why did Dataplex still catch a duplicate issue? The answer is simple - I didn't add a uniqueness test for the transaction_id column in dbt. My focus was on not_null checks, etc., and I missed that one. This is a perfect real-world example of why layering multiple data quality checks matters. The more checks you have, the less likely something slips through!

---

If you want to learn more about Dataplex, stay tuned! I'll be publishing another article soon where you'll learn how to create lakes, zones, glossaries, terms, aspects, and tags, use data insights, and connect everything so your data tells its own story.

---

🌐 If you want to chat about Dataplex or share your experience, find me on LinkedIn!