In week 1, we are supposed to create a network to connect pgadmin to postgres database.

Alexey shows how to do it from CLi, which is the less recommended way if we want to keep the network bridge available for the future.

I did some research online but want able to find a solution. Is it possibile to update docker-compose.yaml with the above specification please?

The README.md under Data ingestion -> Running locally is: URL="https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2021-01.csv"

It should be URL="https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2021-01.parquet"

Raising a PR to add a quick guide to set up dbt on Docker.

Added three files docker-compose.yml, Dockerfile and docker-setup.md for ease of access. Changed the README.md to link it to the setup markdown.

If this is too much clutter then I will just link my repo guide in the README.md and remove all other files from here. Let me know :D

I have tried requesting data from https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2021-01.csv, It seems not available now. Please update the data link from s3. Thanks

A few changes to upload-data.py for week 1:

• cleared the cell outputs (helps with future diffs)
• changed read_csv to read_parquet to reflect the file format change on the TLC Trip Record Data website
• created an interate_df function to emulate the previous read_csv iteration since there is no equivalent for read_parquet
• changed the while loop to a for loop so the StopIteration error is handled automatically

Hi,

I tried to run data_intestrion_gcs_dag from the webserver but the first task stuck at up_for_retry stage. When I tried to check the log this is the only thing I found:

However, I can still run this dag via the Airflow CLI. I'm not sure what's the problem here - I built the image on two different machines, one worked fine, and one got stuck like this.

Hi guys,

Really enjoying the course! Thanks for putting this together.

I couldn't get the data_ingestion_gcs_DAG.py to work, as the NYC taxi data has been updated to parquet format on the website. Here's my workaround, I left in the comments so you can easily see what I changed:

• Changed the dataset_file variable to yellow_tripdata_2021-01.parquet
• Commented out the format_to_parquet function
• Commented out the format_to_parquet_task in the DAG declaration
• Removed format_to_parquet_task from the workflow line at the end of the code

Cheers, Martin

import os
import logging

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

from google.cloud import storage
from airflow.providers.google.cloud.operators.bigquery import BigQueryCreateExternalTableOperator
import pyarrow.csv as pv
import pyarrow.parquet as pq

PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
BUCKET = os.environ.get("GCP_GCS_BUCKET")

dataset_file = "yellow_tripdata_2021-01.parquet"
# dataset_file = "yellow_tripdata_2021-01.csv"
dataset_url = f"https://s3.amazonaws.com/nyc-tlc/trip+data/{dataset_file}"
path_to_local_home = os.environ.get("AIRFLOW_HOME", "/opt/airflow/")
# parquet_file = dataset_file.replace('.csv', '.parquet')
parquet_file = dataset_file
BIGQUERY_DATASET = os.environ.get("BQ_DATASET", 'trips_data_all')


# this is not needed anymore since the file is already in parquet format
# def format_to_parquet(src_file):
#     if not src_file.endswith('.csv'):
#         logging.error("Can only accept source files in CSV format, for the moment")
#         return
#     table = pv.read_csv(src_file)
#     pq.write_table(table, src_file.replace('.csv', '.parquet'))


# NOTE: takes 20 mins, at an upload speed of 800kbps. Faster if your internet has a better upload speed
def upload_to_gcs(bucket, object_name, local_file):
    """
    Ref: https://cloud.google.com/storage/docs/uploading-objects#storage-upload-object-python
    :param bucket: GCS bucket name
    :param object_name: target path & file-name
    :param local_file: source path & file-name
    :return:
    """
    # WORKAROUND to prevent timeout for files > 6 MB on 800 kbps upload speed.
    # (Ref: https://github.com/googleapis/python-storage/issues/74)
    storage.blob._MAX_MULTIPART_SIZE = 5 * 1024 * 1024  # 5 MB
    storage.blob._DEFAULT_CHUNKSIZE = 5 * 1024 * 1024  # 5 MB
    # End of Workaround

    client = storage.Client()
    bucket = client.bucket(bucket)

    blob = bucket.blob(object_name)
    blob.upload_from_filename(local_file)


default_args = {
    "owner": "airflow",
    "start_date": days_ago(1),
    "depends_on_past": False,
    "retries": 1,
}

# NOTE: DAG declaration - using a Context Manager (an implicit way)
with DAG(
    dag_id="data_ingestion_gcs_dag",
    schedule_interval="@daily",
    default_args=default_args,
    catchup=False,
    max_active_runs=1,
    tags=['dtc-de'],
) as dag:

    download_dataset_task = BashOperator(
        task_id="download_dataset_task",
        bash_command=f"curl -sSL {dataset_url} > {path_to_local_home}/{dataset_file}"
    )

    # format_to_parquet_task = PythonOperator(
    #     task_id="format_to_parquet_task",
    #     python_callable=format_to_parquet,
    #     op_kwargs={
    #         "src_file": f"{path_to_local_home}/{dataset_file}",
    #     },
    # )

    # TODO: Homework - research and try XCOM to communicate output values between 2 tasks/operators
    local_to_gcs_task = PythonOperator(
        task_id="local_to_gcs_task",
        python_callable=upload_to_gcs,
        op_kwargs={
            "bucket": BUCKET,
            "object_name": f"raw/{parquet_file}",
            "local_file": f"{path_to_local_home}/{parquet_file}",
        },
    )

    bigquery_external_table_task = BigQueryCreateExternalTableOperator(
        task_id="bigquery_external_table_task",
        table_resource={
            "tableReference": {
                "projectId": PROJECT_ID,
                "datasetId": BIGQUERY_DATASET,
                "tableId": "external_table",
            },
            "externalDataConfiguration": {
                "sourceFormat": "PARQUET",
                "sourceUris": [f"gs://{BUCKET}/raw/{parquet_file}"],
            },
        },
    )

    # download_dataset_task >> format_to_parquet_task >> local_to_gcs_task >> bigquery_external_table_task
    download_dataset_task >> local_to_gcs_task >> bigquery_external_table_task