feat: Extract and materialize new properties from Benefits events

[Scotchester]

Description

Extracts the existing event_properties.(card_category|card_scheme|transit_processor) JSON columns from raw Benefits events data for inclusion in the final mart table. This will allow users to e.g. filter and group on these fields in Metabase.

Resolves cal-itp/benefits#3209

Type of change

• Bug fix (non-breaking change which fixes an issue)
• New feature
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Documentation

How has this been tested?

I am not currently set up to be able to test this. I based the changes off of the example shown in #4166.

Post-merge follow-ups

Document any actions that must be taken post-merge to deploy or otherwise implement the changes in this PR (for example, running a full refresh of some incremental model in dbt). If these actions will take more than a few hours after the merge or if they will be completed by someone other than the PR author, please create a dedicated follow-up issue and link it here to track resolution.

• No action required
• Actions required (specified below)

[github-actions]

Terraform plan in iac/cal-itp-data-infra-staging/composer/us

Plan: 0 to add, 1 to change, 0 to destroy.

Terraform used the selected providers to generate the following execution
plan. Resource actions are indicated with the following symbols:
!~  update in-place

Terraform will perform the following actions:

  # google_composer_environment.calitp-staging-composer will be updated in-place
!~  resource "google_composer_environment" "calitp-staging-composer" {
        id               = "projects/cal-itp-data-infra-staging/locations/us-west2/environments/calitp-staging-composer"
        name             = "calitp-staging-composer"
#        (5 unchanged attributes hidden)

!~      config {
#            (8 unchanged attributes hidden)

!~          software_config {
!~              env_variables            = {
-                   "CALITP_SLACK_URL_KEY"                                 = (sensitive value) -> null
#                    (56 unchanged elements hidden)
                }
#                (6 unchanged attributes hidden)

#                (1 unchanged block hidden)
            }

#            (8 unchanged blocks hidden)
        }

#        (1 unchanged block hidden)
    }

Plan: 0 to add, 1 to change, 0 to destroy.

📝 Plan generated in Plan Terraform for Warehouse and DAG changes #886

[github-actions]

Terraform plan in iac/cal-itp-data-infra-staging/airflow/us

Plan: 0 to add, 2 to change, 0 to destroy.

Terraform used the selected providers to generate the following execution
plan. Resource actions are indicated with the following symbols:
!~  update in-place

Terraform will perform the following actions:

  # google_storage_bucket_object.calitp-staging-composer-dags["models/mart/benefits/_mart_benefits.yml"] will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-staging-composer-dags" {
!~      crc32c              = "/bAzyg==" -> (known after apply)
!~      detect_md5hash      = "Prx7c4IxgDMgw+Po6fnmEA==" -> "different hash"
!~      generation          = 1754938420303891 -> (known after apply)
        id                  = "calitp-staging-composer-data/warehouse/models/mart/benefits/_mart_benefits.yml"
!~      md5hash             = "Prx7c4IxgDMgw+Po6fnmEA==" -> (known after apply)
        name                = "data/warehouse/models/mart/benefits/_mart_benefits.yml"
#        (17 unchanged attributes hidden)
    }

  # google_storage_bucket_object.calitp-staging-composer-dags["models/mart/benefits/fct_benefits_events.sql"] will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-staging-composer-dags" {
!~      crc32c              = "nH3www==" -> (known after apply)
!~      detect_md5hash      = "OBSSkvprb5n9Ku1SrsDS7A==" -> "different hash"
!~      generation          = 1755876750508611 -> (known after apply)
        id                  = "calitp-staging-composer-data/warehouse/models/mart/benefits/fct_benefits_events.sql"
!~      md5hash             = "OBSSkvprb5n9Ku1SrsDS7A==" -> (known after apply)
        name                = "data/warehouse/models/mart/benefits/fct_benefits_events.sql"
#        (17 unchanged attributes hidden)
    }

Plan: 0 to add, 2 to change, 0 to destroy.

📝 Plan generated in Plan Terraform for Warehouse and DAG changes #886

[github-actions]

Terraform plan in iac/cal-itp-data-infra/airflow/us

Plan: 0 to add, 4 to change, 0 to destroy.

Terraform used the selected providers to generate the following execution
plan. Resource actions are indicated with the following symbols:
!~  update in-place

Terraform will perform the following actions:

  # google_storage_bucket_object.calitp-composer-catalog will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-composer-catalog" {
!~      content             = (sensitive value)
!~      crc32c              = "HFLpFA==" -> (known after apply)
!~      detect_md5hash      = "AxlvqGlF0XQytL7X6uBrow==" -> "different hash"
!~      generation          = 1761674676592506 -> (known after apply)
        id                  = "calitp-composer-data/warehouse/target/catalog.json"
!~      md5hash             = "AxlvqGlF0XQytL7X6uBrow==" -> (known after apply)
        name                = "data/warehouse/target/catalog.json"
#        (16 unchanged attributes hidden)
    }

  # google_storage_bucket_object.calitp-composer-dags["models/mart/benefits/_mart_benefits.yml"] will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-composer-dags" {
!~      crc32c              = "/bAzyg==" -> (known after apply)
!~      detect_md5hash      = "Prx7c4IxgDMgw+Po6fnmEA==" -> "different hash"
!~      generation          = 1754938421540880 -> (known after apply)
        id                  = "calitp-composer-data/warehouse/models/mart/benefits/_mart_benefits.yml"
!~      md5hash             = "Prx7c4IxgDMgw+Po6fnmEA==" -> (known after apply)
        name                = "data/warehouse/models/mart/benefits/_mart_benefits.yml"
#        (17 unchanged attributes hidden)
    }

  # google_storage_bucket_object.calitp-composer-dags["models/mart/benefits/fct_benefits_events.sql"] will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-composer-dags" {
!~      crc32c              = "nH3www==" -> (known after apply)
!~      detect_md5hash      = "OBSSkvprb5n9Ku1SrsDS7A==" -> "different hash"
!~      generation          = 1755876748862218 -> (known after apply)
        id                  = "calitp-composer-data/warehouse/models/mart/benefits/fct_benefits_events.sql"
!~      md5hash             = "OBSSkvprb5n9Ku1SrsDS7A==" -> (known after apply)
        name                = "data/warehouse/models/mart/benefits/fct_benefits_events.sql"
#        (17 unchanged attributes hidden)
    }

  # google_storage_bucket_object.calitp-composer-manifest will be updated in-place
!~  resource "google_storage_bucket_object" "calitp-composer-manifest" {
!~      content             = (sensitive value)
!~      crc32c              = "JvfK6A==" -> (known after apply)
!~      detect_md5hash      = "/ncG1pOau+Ob3onRgEJBiQ==" -> "different hash"
!~      generation          = 1761674678163486 -> (known after apply)
        id                  = "calitp-composer-data/warehouse/target/manifest.json"
!~      md5hash             = "/ncG1pOau+Ob3onRgEJBiQ==" -> (known after apply)
        name                = "data/warehouse/target/manifest.json"
#        (16 unchanged attributes hidden)
    }

Plan: 0 to add, 4 to change, 0 to destroy.

📝 Plan generated in Plan Terraform for Warehouse and DAG changes #886

[thekaveman]

I will try to run the new query in the warehouse and post the results, once I can sign in again... Microsoft having some issues today.

[thekaveman]

I adapted the query to be able to run it directly in the Warehouse and see the expected results:

I expect that once we push the latest analytics code to prod with cal-itp/benefits#3228, then we'll start seeing that data in the expected columns as well.

LGTM 👍