<jupyter_start><jupyter_text>dbt Postgres Data Pipeline  Setup the environment Install postgresql<jupyter_code># Install postgresql server
!sudo apt-get -y -qq update
!sudo apt-get -y -qq install postgresql
!sudo service postgresql start

# Setup a password `<PASSWORD>` for username `postgres`
!sudo -u postgres psql -U postgres -c "ALTER USER postgres PASSWORD '<PASSWORD>';"
! lsof -i -P -n | grep -E 'postgres'<jupyter_output>postgres  3539 postgres    7u  IPv4  88798      0t0  TCP 127.0.0.1:5432 (LISTEN)
postgres  3539 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3541 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3542 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3543 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3544 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3545 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994 
postgres  3546 postgres   11u  IPv4  88809      0t0  UDP 127.0.0.1:41994->127.0.0.1:41994<jupyter_text>Install dbt<jupyter_code>!pip install dbt<jupyter_output><empty_output><jupyter_text>Initiate a project<jupyter_code>!dbt init dbt_demo
%cd dbt_demo
!ls -la<jupyter_output>total 671064
drwxr-xr-x 11 root root      4096 Aug 26 08:47 .
drwxr-xr-x  1 root root      4096 Aug 26 08:00 ..
drwxr-xr-x  2 root root      4096 Aug 26 08:00 analysis
drwxr-xr-x  2 root root      4096 Aug 26 08:00 data
drwxr-xr-x  2 root root      4096 Aug 26 08:19 dbt_modules
-rw-r--r--  1 root root       366 Aug 26 08:34 dbt_project.yml
-rw-r--r--  1 root root        28 Aug 26 08:00 .gitignore
drwxr-xr-x  2 root root      4096 Aug 26 08:19 logs
drwxr-xr-x  2 root root      4096 Aug 26 08:00 macros
drwxr-xr-x  4 root root      4096 Aug 26 08:22 models
-rw-r--r--  1 root root       566 Aug 26 08:00 README.md
drwxr-xr-x  2 root root      4096 Aug 26 08:00 snapshots
drwxr-xr-x  4 root root      4096 Aug 26 08:19 target
-rw-r--r--  1 root root     12322 Aug 17  2016 taxi+_zone_lookup.csv
drwxr-xr-x  2 root root      4096 Aug 26 08:00 tests
-rw-r--r--  1 root root 687088084 Aug 14  2019 yellow_tripdata_2019-01.csv<jupyter_text>Load sample data into database<jupyter_code># data source - https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page
!wget https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2019-01.csv
!wget https://s3.amazonaws.com/nyc-tlc/misc/taxi+_zone_lookup.csv
import pandas as pd

yellow_tripdata_2019_df = pd.read_csv('yellow_tripdata_2019-01.csv')
yellow_tripdata_2019_df.head()
yellow_tripdata_2019_df.columns
yellow_tripdata_2019_df = yellow_tripdata_2019_df[['VendorID',
                                                   'tpep_pickup_datetime',
                                                   'tpep_dropoff_datetime',
                                                   'passenger_count',
                                                   'PULocationID',
                                                   'DOLocationID',
                                                   'fare_amount']]

yellow_tripdata_2019_df.columns = ['vendor_id',
                                   'pickup_datetime',
                                   'dropoff_datetime',
                                   'passenger_count',
                                   'pickup_location_id',
                                   'dropoff_location_id',
                                   'fare_amount']
yellow_tripdata_2019_df.info()
taxi_zone_lookup = pd.read_csv('taxi+_zone_lookup.csv')
taxi_zone_lookup.head()
taxi_zone_lookup.columns = ['locationid','borough','zone','service_zone']
taxi_zone_lookup.info()
from sqlalchemy import create_engine
import psycopg2

alchemyEngine = create_engine('postgresql+psycopg2://postgres:postgres@127.0.0.1/postgres', pool_recycle=3600);
postgreSQLConnection = alchemyEngine.connect();

yellow_tripdata_2019_df.sample(100000).to_sql('yellow_tripdata_sample_2019_01', postgreSQLConnection, if_exists='replace');
taxi_zone_lookup.to_sql('taxi_zone_lookup', postgreSQLConnection, if_exists='replace');

postgreSQLConnection.close();<jupyter_output><empty_output><jupyter_text>Configure the project<jupyter_code>!ls ~ -la
!ls ~/.dbt -la
%%writefile ~/.dbt/profiles.yml
default:
  outputs:
    dev:
      type: postgres
      threads: 1
      host: localhost
      port: 5432
      user: postgres
      pass: <PASSWORD>
      dbname: postgres
      schema: public
    prod:
      type: postgres
      threads: 1
      host: localhost
      port: 5432
      user: postgres
      pass: <PASSWORD>
      dbname: postgres
      schema: public
  target: dev
!dbt debug<jupyter_output>Running with dbt=0.20.1
dbt version: 0.20.1
python version: 3.7.11
python path: /usr/bin/python3
os info: Linux-5.4.104+-x86_64-with-Ubuntu-18.04-bionic
Using profiles.yml file at /root/.dbt/profiles.yml
Using dbt_project.yml file at /content/dbt_demo/dbt_project.yml

Configuration:
  profiles.yml file [[32mOK found and valid[0m]
  dbt_project.yml file [[32mOK found and valid[0m]

Required dependencies:
 - git [[32mOK found[0m]

Connection:
  host: localhost
  port: 5432
  user: postgres
  database: postgres
  schema: public
  search_path: None
  keepalives_idle: 0
  sslmode: None
  Connection test: [32mOK connection ok[0m

[0m<jupyter_text>Run dbt sample models<jupyter_code>!apt-get --quiet install tree
!tree ./models
!dbt run<jupyter_output>Running with dbt=0.20.1
Found 2 models, 4 tests, 0 snapshots, 0 analyses, 147 macros, 0 operations, 0 seed files, 0 sources, 0 exposures

08:19:55 | Concurrency: 1 threads (target='dev')
08:19:55 | 
08:19:55 | 1 of 2 START table model public.my_first_dbt_model................... [RUN]
08:19:55 | 1 of 2 OK created table model public.my_first_dbt_model.............. [[32mSELECT 2[0m in 0.10s]
08:19:55 | 2 of 2 START view model public.my_second_dbt_model................... [RUN]
08:19:55 | 2 of 2 OK created view model public.my_second_dbt_model.............. [[32mCREATE VIEW[0m in 0.06s]
08:19:55 | 
08:19:55 | Finished running 1 table model, 1 view model in 0.30s.

[32mCompleted successfully[0m

Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2
[0m<jupyter_text>Create new models StagingThese files are so-called staging models, a pattern commonly used in dbt to prevent access to raw data. Staging models typically simply select from the source data and, if needed, contain some light transformations such as column renamings.<jupyter_code>!mkdir -p ./models/taxi/staging
%%writefile ./models/taxi/staging/schema_staging.yml
version: 2

sources:
  - name: source
    schema: public
    tables:
      - name: taxi_zone_lookup
      - name: yellow_tripdata_sample_2019_01

models:
  - name: stg_taxi_zone_lookup
    description: "A list of all taxi zones with codes in NYC"
    columns:
      - name: locationid
        tests:
          - not_null
      - name: borough
        tests:
          - not_null
      - name: zone
        tests:
          - not_null
      - name: service_zone
        tests:
          - not_null
  - name: stg_taxi_trips
    description: "A reduced version of yellow taxi trip data in NYC"
    columns:
      - name: vendor_id
        tests:
          - not_null
          - accepted_values:
              values: ['1', '2', '4']
      - name: pickup_datetime
        tests:
          - not_null
      - name: dropoff_datetime
        tests:
          - not_null
      - name: passenger_count
        tests:
          - not_null
      - name: pickup_location_id
        tests:
          - not_null
      - name: dropoff_location_id
        tests:
          - not_null
      - name: fare_amount
        tests:
          - not_null
%%writefile ./models/taxi/staging/stg_taxi_trips.sql
select 
    vendor_id,
    pickup_datetime, 
    dropoff_datetime, 
    passenger_count, 
    pickup_location_id, 
    dropoff_location_id, 
    fare_amount
from {{ source('source', 'yellow_tripdata_sample_2019_01') }}
%%writefile ./models/taxi/staging/stg_taxi_zone_lookup.sql
select 
    locationid,
    borough,
    zone,
    service_zone
from {{ source('source', 'taxi_zone_lookup') }}<jupyter_output>Writing ./models/taxi/staging/stg_taxi_zone_lookup.sql<jupyter_text>Create a New ModelWe will now create our first dbt model, which combines data from the two staging models. Let's assume we want to write a query to join the staging tables on the location ID fields and add the actual location names to the pickup and dropoff locations of the taxi ride data.<jupyter_code>%%writefile ./models/taxi/trips_with_borough_name.sql
select
    t.vendor_id,
    t.pickup_datetime,
    t.dropoff_datetime,
    z1.borough as pickup_borough,
    z2.borough as dropoff_borough,
    t.passenger_count,
    t.fare_amount
from {{ ref('stg_taxi_trips') }} t
left join {{ ref('stg_taxi_zone_lookup') }} z1
on t.pickup_location_id = z1.locationid
left join {{ ref('stg_taxi_zone_lookup') }} z2
on t.dropoff_location_id = z2.locationid<jupyter_output>Writing ./models/taxi/trips_with_borough_name.sql<jupyter_text>Create the schema<jupyter_code>%%writefile ./models/taxi/schema.yml
version: 2

models:
  - name: trips_with_borough_name
    description: "Combines taxi rides with the borough names for pickup and dropoff locations."
    columns:
      - name: vendor_id
      - name: pickup_datetime
      - name: dropoff_datetime
      - name: pickup_borough
      - name: dropoff_borough
      - name: passenger_count
      - name: fare_amount<jupyter_output>Writing ./models/taxi/schema.yml<jupyter_text>Configure the dbt_project.yml file This file defines how the data in the models under taxi will be materialized. Modify the section to replace example with our new taxi directory:<jupyter_code>!cat dbt_project.yml
%%writefile dbt_project.yml
name: 'my_new_project'
version: '1.0.0'
config-version: 2
profile: 'default'
source-paths: ["models"]
analysis-paths: ["analysis"]
test-paths: ["tests"]
data-paths: ["data"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
target-path: "target"
clean-targets:
    - "target"
    - "dbt_modules"
models:
  my_new_project:
      taxi:
          materialized: view<jupyter_output>Overwriting dbt_project.yml<jupyter_text>Run the new pipeline<jupyter_code>!dbt run<jupyter_output>Running with dbt=0.20.1
Found 5 models, 16 tests, 0 snapshots, 0 analyses, 147 macros, 0 operations, 0 seed files, 2 sources, 0 exposures

09:16:23 | Concurrency: 1 threads (target='dev')
09:16:23 | 
09:16:23 | 1 of 5 START table model public.my_first_dbt_model................... [RUN]
09:16:23 | 1 of 5 OK created table model public.my_first_dbt_model.............. [[32mSELECT 2[0m in 0.09s]
09:16:23 | 2 of 5 START view model public.stg_taxi_trips........................ [RUN]
09:16:23 | 2 of 5 OK created view model public.stg_taxi_trips................... [[32mCREATE VIEW[0m in 0.06s]
09:16:23 | 3 of 5 START view model public.stg_taxi_zone_lookup.................. [RUN]
09:16:23 | 3 of 5 OK created view model public.stg_taxi_zone_lookup............. [[32mCREATE VIEW[0m in 0.03s]
09:16:23 | 4 of 5 START view model public.my_second_dbt_model................... [RUN]
09:16:23 | 4 of 5 OK created view model public.my_second_dbt_model.............. [[32mCREATE VIEW[0m in 0.04s]
09:[...]<jupyter_text>Running tests Run the staging schema tests<jupyter_code>!dbt test -m stg_taxi_trips stg_taxi_zone_lookup<jupyter_output>Running with dbt=0.20.1
Found 5 models, 16 tests, 0 snapshots, 0 analyses, 147 macros, 0 operations, 0 seed files, 2 sources, 0 exposures

09:18:04 | Concurrency: 1 threads (target='dev')
09:18:04 | 
09:18:04 | 1 of 12 START test accepted_values_stg_taxi_trips_vendor_id__1__2__4. [RUN]
09:18:04 | 1 of 12 PASS accepted_values_stg_taxi_trips_vendor_id__1__2__4....... [[32mPASS[0m in 0.08s]
09:18:04 | 2 of 12 START test not_null_stg_taxi_trips_dropoff_datetime.......... [RUN]
09:18:04 | 2 of 12 PASS not_null_stg_taxi_trips_dropoff_datetime................ [[32mPASS[0m in 0.03s]
09:18:04 | 3 of 12 START test not_null_stg_taxi_trips_dropoff_location_id....... [RUN]
09:18:04 | 3 of 12 PASS not_null_stg_taxi_trips_dropoff_location_id............. [[32mPASS[0m in 0.03s]
09:18:04 | 4 of 12 START test not_null_stg_taxi_trips_fare_amount............... [RUN]
09:18:05 | 4 of 12 PASS not_null_stg_taxi_trips_fare_amount..................... [[32mPASS[0m in 0.03s]
09:18:05 | 5 of 12 START tes[...]<jupyter_text>Correct the failure<jupyter_code>%%writefile ./models/taxi/staging/stg_taxi_zone_lookup.sql
select 
    locationid,
    borough,
    zone,
    service_zone
from {{ source('source', 'taxi_zone_lookup') }} lk
where lk.zone is not null
and lk.service_zone is not null
!dbt run
!dbt test -m stg_taxi_trips stg_taxi_zone_lookup<jupyter_output>Running with dbt=0.20.1
Found 5 models, 16 tests, 0 snapshots, 0 analyses, 147 macros, 0 operations, 0 seed files, 2 sources, 0 exposures

09:24:33 | Concurrency: 1 threads (target='dev')
09:24:33 | 
09:24:33 | 1 of 12 START test accepted_values_stg_taxi_trips_vendor_id__1__2__4. [RUN]
09:24:33 | 1 of 12 PASS accepted_values_stg_taxi_trips_vendor_id__1__2__4....... [[32mPASS[0m in 0.06s]
09:24:33 | 2 of 12 START test not_null_stg_taxi_trips_dropoff_datetime.......... [RUN]
09:24:33 | 2 of 12 PASS not_null_stg_taxi_trips_dropoff_datetime................ [[32mPASS[0m in 0.03s]
09:24:33 | 3 of 12 START test not_null_stg_taxi_trips_dropoff_location_id....... [RUN]
09:24:33 | 3 of 12 PASS not_null_stg_taxi_trips_dropoff_location_id............. [[32mPASS[0m in 0.03s]
09:24:33 | 4 of 12 START test not_null_stg_taxi_trips_fare_amount............... [RUN]
09:24:33 | 4 of 12 PASS not_null_stg_taxi_trips_fare_amount..................... [[32mPASS[0m in 0.03s]
09:24:33 | 5 of 12 START tes[...]
{'fraction_non_alphanumeric': 0.12117494226327945, 'fraction_numerical': 0.080614896073903, 'mean_word_length': 2.934412265758092, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 53, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 16, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '554711', 'license': 'Apache-2.0', 'n_tokens_mistral': 5725, 'n_tokens_neox': 4914, 'n_words': 1261, 'path': '_docs/nbs/2022-02-01-dbt.ipynb', 'repo_name': 'sparsh-ai/recohut', 'size': '13848'}