Transformations with SQL (Part 1/3)

Transformations with SQL (Part 1/3)

Overview

This tutorial will describe how to integrate SQL based transformations with Airbyte syncs using plain SQL queries.
This is the first part of ELT tutorial. The second part goes deeper with Transformations with dbt and then wrap-up with a third part on Transformations with Airbyte.

(Examples outputs are updated with Airbyte version 0.23.0-alpha from May 2021)

First transformation step: Normalization

At its core, Airbyte is geared to handle the EL (Extract Load) steps of an ELT process. These steps can also be referred in Airbyte's dialect as "Source" and "Destination".
However, this is actually producing a table in the destination with a JSON blob column... For the typical analytics use case, you probably want this json blob normalized so that each field is its own column.
So, after EL, comes the T (transformation) and the first T step that Airbyte actually applies on top of the extracted data is called "Normalization". You can find more information about it here.
Airbyte runs this step before handing the final data over to other tools that will manage further transformation down the line.
To summarize, we can represent the ELT process in the diagram below. These are steps that happens between your "Source Database or API" and the final "Replicated Tables" with examples of implementation underneath:
Anyway, it is possible to short-circuit this process (no vendor lock-in) and handle it yourself by turning this option off in the destination settings page.
This could be useful if:
  1. 1.
    You have a use-case not related to analytics that could be handled with data in its raw JSON format.
  2. 2.
    You can implement your own transformer. For example, you could write them in a different language, create them in an analytics engine like Spark, or use a transformation tool such as dbt or Dataform.
  3. 3.
    You want to customize and change how the data is normalized with your own queries.
In order to do so, we will now describe how you can leverage the basic normalization outputs that Airbyte generates to build your own transformations if you don't want to start from scratch.
Note: We will rely on docker commands that we've gone over as part of another Tutorial on Exploring Docker Volumes.

(Optional) Configure some Covid (data) source and Postgres destinations

If you have sources and destinations already setup on your deployment, you can skip to the next section.
For the sake of this tutorial, let's create some source and destination as an example that we can refer to afterward. We'll be using a file accessible from a public API, so you can easily reproduce this setup:
1
Here are some examples of public API CSV:
2
https://storage.googleapis.com/covid19-open-data/v2/latest/epidemiology.csv
Copied!
And a local Postgres Database, making sure that "Basic normalization" is enabled:
After setting up the connectors, we can trigger the sync and study the logs:
Notice that the process ran in the /tmp/workspace/5/0 folder.

Identify Workspace ID with Normalize steps

If you went through the previous setup of source/destination section and run a sync, you were able to identify which workspace was used, let's define some environment variables to remember this:
1
NORMALIZE_WORKSPACE="5/0/"
Copied!
Or if you want to find any folder where the normalize step was run:
1
# find automatically latest workspace where normalization was run
2
NORMALIZE_WORKSPACE=`docker run --rm -i -v airbyte_workspace:/data busybox find /data -path "*normalize/models*" | sed -E "s;/data/([0-9]+/[0-9]+/)normalize/.*;\1;g" | sort | uniq | tail -n 1`
Copied!

Export Plain SQL files

Airbyte is internally using a specialized tool for handling transformations called dbt.
The Airbyte Python module reads the destination_catalog.json file and generates dbt code responsible for interpreting and transforming the raw data.
The final output of dbt is producing SQL files that can be run on top of the destination that you selected.
Therefore, it is possible to extract these SQL files, modify them and run it yourself manually outside Airbyte!
You would be able to find these at the following location inside the server's docker container:
1
/tmp/workspace/${NORMALIZE_WORKSPACE}/build/run/airbyte_utils/models/generated/airbyte_tables/<schema>/<your_table_name>.sql
Copied!
In order to extract them, you can run:
1
#!/usr/bin/env bash
2
docker cp airbyte-server:/tmp/workspace/${NORMALIZE_WORKSPACE}/build/run/airbyte_utils/models/generated/ models/
3
4
find models
Copied!
Example Output:
1
models/airbyte_tables/quarantine/covid_epidemiology_f11.sql
Copied!
Let's inspect the generated SQL file by running:
1
cat models/**/covid_epidemiology*.sql
Copied!
Example Output:
1
create table "postgres".quarantine."covid_epidemiology_f11__dbt_tmp"
2
as (
3
4
with __dbt__CTE__covid_epidemiology_ab1_558 as (
5
6
-- SQL model to parse JSON blob stored in a single column and extract into separated field columns as described by the JSON Schema
7
select
8
jsonb_extract_path_text(_airbyte_data, 'key') as "key",
9
jsonb_extract_path_text(_airbyte_data, 'date') as "date",
10
jsonb_extract_path_text(_airbyte_data, 'new_tested') as new_tested,
11
jsonb_extract_path_text(_airbyte_data, 'new_deceased') as new_deceased,
12
jsonb_extract_path_text(_airbyte_data, 'total_tested') as total_tested,
13
jsonb_extract_path_text(_airbyte_data, 'new_confirmed') as new_confirmed,
14
jsonb_extract_path_text(_airbyte_data, 'new_recovered') as new_recovered,
15
jsonb_extract_path_text(_airbyte_data, 'total_deceased') as total_deceased,
16
jsonb_extract_path_text(_airbyte_data, 'total_confirmed') as total_confirmed,
17
jsonb_extract_path_text(_airbyte_data, 'total_recovered') as total_recovered,
18
_airbyte_emitted_at
19
from "postgres".quarantine._airbyte_raw_covid_epidemiology
20
-- covid_epidemiology
21
), __dbt__CTE__covid_epidemiology_ab2_558 as (
22
23
-- SQL model to cast each column to its adequate SQL type converted from the JSON schema type
24
select
25
cast("key" as
26
varchar
27
) as "key",
28
cast("date" as
29
varchar
30
) as "date",
31
cast(new_tested as
32
float
33
) as new_tested,
34
cast(new_deceased as
35
float
36
) as new_deceased,
37
cast(total_tested as
38
float
39
) as total_tested,
40
cast(new_confirmed as
41
float
42
) as new_confirmed,
43
cast(new_recovered as
44
float
45
) as new_recovered,
46
cast(total_deceased as
47
float
48
) as total_deceased,
49
cast(total_confirmed as
50
float
51
) as total_confirmed,
52
cast(total_recovered as
53
float
54
) as total_recovered,
55
_airbyte_emitted_at
56
from __dbt__CTE__covid_epidemiology_ab1_558
57
-- covid_epidemiology
58
), __dbt__CTE__covid_epidemiology_ab3_558 as (
59
60
-- SQL model to build a hash column based on the values of this record
61
select
62
*,
63
md5(cast(
64
65
coalesce(cast("key" as
66
varchar
67
), '') || '-' || coalesce(cast("date" as
68
varchar
69
), '') || '-' || coalesce(cast(new_tested as
70
varchar
71
), '') || '-' || coalesce(cast(new_deceased as
72
varchar
73
), '') || '-' || coalesce(cast(total_tested as
74
varchar
75
), '') || '-' || coalesce(cast(new_confirmed as
76
varchar
77
), '') || '-' || coalesce(cast(new_recovered as
78
varchar
79
), '') || '-' || coalesce(cast(total_deceased as
80
varchar
81
), '') || '-' || coalesce(cast(total_confirmed as
82
varchar
83
), '') || '-' || coalesce(cast(total_recovered as
84
varchar
85
), '')
86
87
as
88
varchar
89
)) as _airbyte_covid_epidemiology_hashid
90
from __dbt__CTE__covid_epidemiology_ab2_558
91
-- covid_epidemiology
92
)-- Final base SQL model
93
select
94
"key",
95
"date",
96
new_tested,
97
new_deceased,
98
total_tested,
99
new_confirmed,
100
new_recovered,
101
total_deceased,
102
total_confirmed,
103
total_recovered,
104
_airbyte_emitted_at,
105
_airbyte_covid_epidemiology_hashid
106
from __dbt__CTE__covid_epidemiology_ab3_558
107
-- covid_epidemiology from "postgres".quarantine._airbyte_raw_covid_epidemiology
108
);
Copied!

Simple SQL Query

We could simplify the SQL query by removing some parts that may be unnecessary for your current usage (such as generating a md5 column; Why exactly would I want to use that?!).
It would turn into a simpler query:
1
create table "postgres"."public"."covid_epidemiology"
2
as (
3
select
4
_airbyte_emitted_at,
5
(current_timestamp at time zone 'utc')::timestamp as _airbyte_normalized_at,
6
7
cast(jsonb_extract_path_text("_airbyte_data",'key') as varchar) as "key",
8
cast(jsonb_extract_path_text("_airbyte_data",'date') as varchar) as "date",
9
cast(jsonb_extract_path_text("_airbyte_data",'new_tested') as float) as new_tested,
10
cast(jsonb_extract_path_text("_airbyte_data",'new_deceased') as float) as new_deceased,
11
cast(jsonb_extract_path_text("_airbyte_data",'total_tested') as float) as total_tested,
12
cast(jsonb_extract_path_text("_airbyte_data",'new_confirmed') as float) as new_confirmed,
13
cast(jsonb_extract_path_text("_airbyte_data",'new_recovered') as float) as new_recovered,
14
cast(jsonb_extract_path_text("_airbyte_data",'total_deceased') as float) as total_deceased,
15
cast(jsonb_extract_path_text("_airbyte_data",'total_confirmed') as float) as total_confirmed,
16
cast(jsonb_extract_path_text("_airbyte_data",'total_recovered') as float) as total_recovered
17
from "postgres".public._airbyte_raw_covid_epidemiology
18
);
Copied!

Customize SQL Query

Feel free to:
  • Rename the columns as you desire
    • avoiding using keywords such as "key" or "date"
  • You can tweak the column data type if the ones generated by Airbyte are not the ones you favor
    • For example, let's use Integer instead of Float for the number of Covid cases...
  • Add deduplicating logic
    • if you can identify which columns to use as Primary Keys
      (since airbyte isn't able to detect those automatically yet...)
    • (Note: actually I am not even sure if I can tell the proper primary key in this dataset...)
  • Create a View (or materialized views) instead of a Table.
  • etc
1
create view "postgres"."public"."covid_epidemiology" as (
2
with parse_json_cte as (
3
select
4
_airbyte_emitted_at,
5
6
cast(jsonb_extract_path_text("_airbyte_data",'key') as varchar) as id,
7
cast(jsonb_extract_path_text("_airbyte_data",'date') as varchar) as updated_at,
8
cast(jsonb_extract_path_text("_airbyte_data",'new_tested') as float) as new_tested,
9
cast(jsonb_extract_path_text("_airbyte_data",'new_deceased') as float) as new_deceased,
10
cast(jsonb_extract_path_text("_airbyte_data",'total_tested') as float) as total_tested,
11
cast(jsonb_extract_path_text("_airbyte_data",'new_confirmed') as float) as new_confirmed,
12
cast(jsonb_extract_path_text("_airbyte_data",'new_recovered') as float) as new_recovered,
13
cast(jsonb_extract_path_text("_airbyte_data",'total_deceased') as float) as total_deceased,
14
cast(jsonb_extract_path_text("_airbyte_data",'total_confirmed') as float) as total_confirmed,
15
cast(jsonb_extract_path_text("_airbyte_data",'total_recovered') as float) as total_recovered
16
from "postgres".public._airbyte_raw_covid_epidemiology
17
),
18
cte as (
19
select
20
*,
21
row_number() over (
22
partition by id
23
order by updated_at desc
24
) as row_num
25
from parse_json_cte
26
)
27
select
28
substring(id, 1, 2) as id, -- Probably not the right way to identify the primary key in this dataset...
29
updated_at,
30
_airbyte_emitted_at,
31
32
case when new_tested = 'NaN' then 0 else cast(new_tested as integer) end as new_tested,
33
case when new_deceased = 'NaN' then 0 else cast(new_deceased as integer) end as new_deceased,
34
case when total_tested = 'NaN' then 0 else cast(total_tested as integer) end as total_tested,
35
case when new_confirmed = 'NaN' then 0 else cast(new_confirmed as integer) end as new_confirmed,
36
case when new_recovered = 'NaN' then 0 else cast(new_recovered as integer) end as new_recovered,
37
case when total_deceased = 'NaN' then 0 else cast(total_deceased as integer) end as total_deceased,
38
case when total_confirmed = 'NaN' then 0 else cast(total_confirmed as integer) end as total_confirmed,
39
case when total_recovered = 'NaN' then 0 else cast(total_recovered as integer) end as total_recovered
40
from cte
41
where row_num = 1
42
);
Copied!
Then you can run in your preferred SQL editor or tool!
If you are familiar with dbt or want to learn more about it, you can continue with the following tutorial using dbt...
Last modified 2mo ago