GitLab Duo Agent平台如何变革DataOps

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

SELECT 
    table_name,
    column_name,
    data_type,
    is_nullable,
    CASE 
        WHEN is_nullable = 'NO' THEN 'PRIMARY_KEY'
        ELSE NULL 
    END as key_type
FROM raw.information_schema.columns
WHERE table_schema = 'REDDIT_ADS'
ORDER BY table_name, ordinal_position;

1
2
3
4
5
6
7
8

Create dbt models for all the tables in the file structure.csv.

I want to have the source models created, with a filter that dedupes the data based on the primary key. Create these in a new folder reddit_ads.
I want to have workspace models created and store these in the workspace_marketing schema.

Take this MR as example: [I've referenced to previous source implementation]. Here is the same done for Source A, but now it needs to be done for Reddit Ads. 

Please check the dbt style guide when creating the code: https://handbook.gitlab.com/handbook/enterprise-data/platform/dbt-guide/

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

transform/snowflake-dbt/
├── dbt_project.yml                                                    [MODIFIED]
└── models/
    ├── sources/
    │   └── reddit_ads/
    │       ├── reddit_ads_ad_group_source.sql                        [NEW]
    │       ├── reddit_ads_ad_source.sql                              [NEW]
    │       ├── reddit_ads_business_account_source.sql                [NEW]
    │       ├── reddit_ads_campaign_source.sql                        [NEW]
    │       ├── reddit_ads_custom_audience_history_source.sql         [NEW]
    │       ├── reddit_ads_geolocation_source.sql                     [NEW]
    │       ├── reddit_ads_interest_source.sql                        [NEW]
    │       ├── reddit_ads_targeting_community_source.sql             [NEW]
    │       ├── reddit_ads_targeting_custom_audience_source.sql       [NEW]
    │       ├── reddit_ads_targeting_device_source.sql                [NEW]
    │       ├── reddit_ads_targeting_geolocation_source.sql           [NEW]
    │       ├── reddit_ads_targeting_interest_source.sql              [NEW]
    │       ├── reddit_ads_time_zone_source.sql                       [NEW]
    │       ├── schema.yml                                            [NEW]
    │       └── sources.yml                                           [NEW]
    └── workspaces/
        └── workspace_marketing/
            └── reddit_ads/
                ├── schema.yml                                        [NEW]
                ├── wk_reddit_ads_ad.sql                              [NEW]
                ├── wk_reddit_ads_ad_group.sql                        [NEW]
                ├── wk_reddit_ads_business_account.sql                [NEW]
                ├── wk_reddit_ads_campaign.sql                        [NEW]
                ├── wk_reddit_ads_custom_audience_history.sql         [NEW]
                ├── wk_reddit_ads_geolocation.sql                     [NEW]
                ├── wk_reddit_ads_interest.sql                        [NEW]
                ├── wk_reddit_ads_targeting_community.sql             [NEW]
                ├── wk_reddit_ads_targeting_custom_audience.sql       [NEW]
                ├── wk_reddit_ads_targeting_device.sql                [NEW]
                ├── wk_reddit_ads_targeting_geolocation.sql           [NEW]
                └── wk_reddit_ads_time_zone.sql                       [NEW]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

WITH source AS (
  SELECT *
  FROM {{ source('reddit_ads','time_zone') }}
  QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY _fivetran_synced DESC) = 1
),

renamed AS (
  SELECT
    id::VARCHAR                               AS time_zone_id,
    code::VARCHAR                             AS time_zone_code,
    dst_offset::NUMBER                        AS time_zone_dst_offset,
    is_dst_active::BOOLEAN                    AS is_time_zone_dst_active,
    name::VARCHAR                             AS time_zone_name,
    offset::NUMBER                            AS time_zone_offset,
    _fivetran_synced::TIMESTAMP               AS fivetran_synced_at
  FROM source
)

SELECT * FROM renamed

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

models:
  - name: reddit_ads_time_zone_source
    description: Time zone data from Reddit Ads system
    columns:
      - name: time_zone_id
        description: Unique identifier for time zone records
        data_tests:
          - unique
          - not_null
      - name: time_zone_code
        description: Code for the time zone
      - name: time_zone_dst_offset
        description: Daylight saving time offset for the time zone
      - name: is_time_zone_dst_active
        description: Flag indicating if daylight saving time is active
      - name: time_zone_name
        description: Name of the time zone
      - name: time_zone_offset
        description: Offset for the time zone
      - name: fivetran_synced_at
        description: Timestamp when the record was last synced by Fivetran

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

sources:
  - name: reddit_ads
    database: RAW
    schema: reddit_ads
    loaded_at_field: _fivetran_synced
    loader: fivetran
    description: Reddit Ads data

    quoting:
      database: true
      schema: false
      identifier: false

    tables:
      - name: time_zone

1
2
3
4
5
6

WITH source AS (
  SELECT *
  FROM {{ ref('reddit_ads_time_zone_source') }}
)

SELECT * FROM source

1

Create a test query to test the row counts between the raw layer and the workspace layer. Keep in mind that we do deduplication, so we can compare both using distinct on the primary keys.