Workflows · From Tasks

From tasks handle all data imports to your data lake. You can configure the items below. Click on the name for more info.

• task
• extract
• file
• transforms
• schema
• deduplicate
• load
• validate

Extract

The `extract` part of the configuration considers the extract part of a task. The vocabulary of the datasource is leading in this part. Will Workflows talk about fields and records, it will use columns and rows here if the datasource uses those.

All from tasks come with pre-formatted schemas where possible. Examples of these are from_facebook or from_xandr. Tasks where a pre-formatted schema is not possible, we give you the opportunity to create one. Examples of these tasks are from_imap (which imports mail attachments) or from_aws_s3 (which imports files from Amazon S3)

New task types are added to Workflows on a monthly basis. Currently Workflows supports the following from tasks:

1. from_appfigures
2. from_apple_app_store_connect
3. from_aws_s3
4. from_bigquery
5. from_bing
6. from_bluesky
7. from_dcm
8. from_dcbm
9. from_facebook
10. from_ftp
11. from_google_ads
12. from_google_analytics
13. from_google_analytics_management
14. from_google_drive
15. from_google_search_console
16. from_google_sheets
17. from_imap
18. from_linkedin
19. from_looker
20. from_url
21. from_salesforce
22. from_snowflake
23. from_x
24. from_xandr

File

The file part of the configuration contains the definition of the data file being downloaded. Use for from task where the format is not pre-defined. E.g. with from_aws_s3 to download a file.

Usage

file:
    type: txt
    transforms:
        - type: uncompress
    delimiter: ;
    encoding: UTF-8
    text_enclosure: '"'
    new_line: '\n'
    escape_char: '\'
    has_header_row: yes
    skip_rows: 9
    skip_bottom_rows: 8

Properties

transforms

The `transforms` part of the configuration enables you to manipulate data before you load it in your data lake. You have to add all transform as an array.

Example usage

transforms:
    # Field transforms
    - type: fields_to_snake_case
    - type: fields_to_lower_case
    - type: fields_regexp_and_replace
        search: '_'
        replace: ''

Transform types

Below are the types of transforms. You have to add them as an array. Below is an explanation per transform on how to use it exactly.

• add_current_date_field
• convert_type
• rename_field
• copy_field
• drop_fields
• encrypt
• fields_search_and_replace
• fields_to_snake_case
• fields_to_lower_case
• fields_regexp_and_replace
• filter_records
• find_all
• find_first
• regexp_and_replace
• replace_with_null
• search_and_replace
• strptime

transform: add_current_date_field

Adds the current date (trigger_date) as a field.

transform: convert_type

Converts the type of a field. Only useful with BigQuery and no schema is set.

transform: rename_field

Rename a field

transform: copy_field

Duplicate a field.

transform: drop_fields

Drops fields.

transform: encrypt

Encrypt field content. Support for several hash algorithms.

transform: fields_search_and_replace

Search and replace a field name.

transform: fields_to_snake_case

Transform field names to snake_case. E.g. OneSecondBefore to one_second_before. Also replaces the field names in a schema.

transform: fields_to_lower_case

Transform field names to lower case. E.g. ONESECONDBEFORE to onesecondbefore. Also replaces the field names in a schema.

transform: fields_regexp_and_replace

Regular expression replacement of field names.

transform: filter_records

Filter records from incoming dataset.

transform: find_all

Find all occurrences of a regular expression and return an array of matches. Based on Python's re.findall

transform: find_first

Find the first occurrence of a regular expression and the match. Based on Python's re.findall

transform: regexp_and_replace

Regular expression replacement of field data.

transform: replace_with_null

Conditionally replace field data with null value.

transform: search_and_replace

Search and replace a substring in field data.

transform: strptime

Parse a field value as a time string. Uses Python's datetime.strptime.

Schema

Don't use a schema for the API's that use pre-formatted schemas. The extract section of the task type you chose contains this information. The `schema` part of the configuration is used to create the destination table. Obliged for Snowflake, but optional for BigQuery.

Example usage

schema:
    description: You can put a table description here
    time_partitioning:
        field: date_field
        type: day
        require_partition_filter: yes
    fields:
        - name: date_field
          type: DATE
          mode: REQUIRED
          description: Description for date_field will be visible as field comment. Field will be used for partitioning in BigQuery and clustering in Snowflake.
        - name: some_string_field
          type: STRING
          mode: NULLABLE
          description: Description for some_string_field will be visible as field comment.
        - name: some_int_field
          type: INTEGER
          mode: NULLABLE
          description: Description for some_int_field will be visible as field comment.
        - name: some_repeated_field
          type: OBJECT
          mode: REPEATED
          description: Description for some_int_field will be visible as field comment.
          fields:
            - name: date_field
                type: DATE
                mode: REQUIRED
                description: Description for date_field will be visible as field comment. Field will be used for partitioning in BigQuery and clustering in Snowflake.
            - name: some_string_field
                type: STRING
                mode: NULLABLE
                description: Description for some_string_field will be visible as field comment.

Properties of fields

Obliged. Array of table fields.

Properties of time_partitioning

Optional. Time partitioning considers either clustering (for Snowflake) or time partitioning (for BigQuery).

Other properties

Extra properties that can be added during table creation.

Deduplicate

Optional. Deduplicate the destination table to make sure you don't import the same data twice. Configure once, deduplicates automatically. If no deduplication is added, all imported data will be added to the destination table.

You can deduplicate a storage or table destination. Click on a deduplicate type for an explanation.

• table
• storage

deduplicate: storage

Below are the types of deduplicate. Below is an explanation per deduplicate type.

• none
• delete_all

Example usage

deduplicate:
    type: delete_all

deduplicate storage: none

Default setting. No deduplication. Adds all files to the destination storage location. Overwrites if the source files have the same name as the destination files.

deduplicate storage: delete_all

Deletes all files in the destination storage location.

deduplicate: table

Below are the types of deduplicate. Below is an explanation per deduplicate type.

• append
• drop
• truncate
• date_range
• keys

Example usage

deduplicate:
    type: date_range
    field: booking_date

deduplicate: append

Default setting. No deduplication. Appends all data to the destination table.

deduplicate: drop

Preferred over type=truncate. Drops destination table and recreates the table (with latest fields and descriptions) before importing new data.

deduplicate: truncate

Prefer type=drop over truncate. Truncates (or creates if not exists) destination table before importing new data.

deduplicate: date_range

Deletes the date range from deduplicate.start_date (or task.start_date) till deduplicate.end_date (or deduplicate.end_date) (inclusive) before importing new data. Currently only supports deletion by days. Contact support if you want a different timeframe, e.g. hour.

deduplicate: keys

Basically a merge of the imported dataset into the destination table. First deletes all records in the destination table if the key(s) value(s) are present in the imported dataset. Then appends the imported dataset.

Load

Optional. Configures where in the data lake the imported data should be stored.

Load types

Below are the types of load. Click on a load type for an explanation.

• database (BigQuery)
• database (Snowflake)
• storage

load: database (BigQuery)

Load imported data into a BigQuery table.

Example usage

Example of loading the data files to table `my-data-lake.osb_raw.weather-data`

load:
    destination: database
    conn_id: google_cloud_default
    project_id: my-data-lake
    dataset_id: osb_raw
    table_id: weather-data

load: database (Snowflake)

Load imported data into a Snowflake table.

Example usage

Example of loading the data files to table PRODUCTION.OSB_RAW.WEATHER_DATA

load:
    destination: database
    conn_id: snowflake
    database: PRODUCTION
    schema: OSB_RAW
    table: WEATHER_DATA

load: storage

Example usage

Example of loading the data files to s3://my-test-bucket/some/folder

load:
    destination: storage
    conn_id: s3
    bucket: my-test-bucket
    folder: some/folder

Load imported data to Amazon S3, Google Cloud Storage or Azure Blob Storage (in beta).

validate

The `validate` part of the configuration enables you to check the data you just imported. If the imported data does not pass the validation an error will be generated.

Validation types

Below are the validation types. You have to add each validation as an array. Below is an explanation per validation on how to use it exactly.

• special property: query
• number
• total_rows
• yesno

Special property: query

Set this property to the query you want to execute. Make sure the resultset is flattened (only rows and columns, not nested or repeated columns). The query property is obliged for the first validation, but optional for all subsequent validations. If you choose not to add query to the subsequent validations, Workflows will use the same resultset for the validation until it hits another query property. If you add a query property to any of the subsequent validations, Workflows will execute the query, and use the resultset in that validation and the subsequent validations until it hits a another query property again. You are allowed to use SQL Templating.

Validate: number

Validates a number field.

Example usage

The examples validates if the value of all rows in the resultset of column lines is greater than (>) the value (1000). The example uses SQL Templating, hence the {{ and }}.

validate:
    - query: |
              SELECT my_segment, COUNT(1) AS lines
              FROM `{{ load.project_id }}.{{ load.dataset_id }}.{{ load.table_id }}`
              GROUP BY 1
      type: number
      operator: '>'
      value: 1000
      field: lines

Validate: total_rows

Validates the number of rows of a resultset.

Example usage

The examples validates if the amount of rows equals (==) eight. lines is greater than (>) the value (1000). The example uses SQL Templating, hence the {{ and }}.

validate:
    - query: |
              SELECT my_segment, COUNT(1) AS transactions
              FROM `{{ load.project_id }}.{{ load.dataset_id }}.{{ load.table_id }}`
              GROUP BY 1
      type: total_rows
      operator: '=='
      value: 8

Validate: yesno

Validates if the value of a field is yes (True) in all rows.

Example usage

The examples validates if the value of field has_enough_transactions is yes (True). The example uses SQL Templating, hence the {{ and }}.

validate:
    - query: |
              SELECT COUNT(1) > 1000 AS has_enough_transactions
              FROM `{{ load.project_id }}.{{ load.dataset_id }}.{{ load.table_id }}`
      type: yesno
      field: has_enough_transactions