cloud-dwh-3.py

load_us_cities_demo = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_us_cities_demo',
    bucket = gs_bucket,
    source_objects = ['cities/us-cities-demographics.csv'],
    destination_project_dataset_table = f'{project_id}:{staging_dataset}.us_cities_demo',
    schema_object = 'cities/us_cities_demo.json',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    field_delimiter=';',
    skip_leading_rows = 1
)

load_airports = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_airports',
    bucket = gs_bucket,
    source_objects = ['airports/airport-codes_csv.csv'],
    destination_project_dataset_table = f'{project_id}:{staging_dataset}.airport_codes',
    schema_object = 'airports/airport_codes.json',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    skip_leading_rows = 1
)

load_weather = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_weather',
    bucket = gs_bucket,
    source_objects = ['weather/GlobalLandTemperaturesByCity.csv'],
    destination_project_dataset_table = f'{project_id}:{staging_dataset}.temperature_by_city',
    schema_object = 'weather/temperature_by_city.json',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    skip_leading_rows = 1
)

load_immigration_data = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_immigration_data',
    bucket = gs_bucket,
    source_objects = ['immigration_data/*.parquet'],
    destination_project_dataset_table = f'{project_id}:{staging_dataset}.immigration_data',
    source_format = 'parquet',
    write_disposition='WRITE_TRUNCATE',
    skip_leading_rows = 1,
    autodetect = True
)

load_country = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_country',
    bucket = gs_bucket,
    source_objects = ['master_data/I94CIT_I94RES.csv'],
    destination_project_dataset_table = f'{project_id}:{dwh_dataset}.D_COUNTRY',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    skip_leading_rows = 1,
    schema_fields=[
        {'name': 'COUNTRY_ID', 'type': 'NUMERIC', 'mode': 'NULLABLE'},
        {'name': 'COUNTRY_NAME', 'type': 'STRING', 'mode': 'NULLABLE'},
    ]
)

load_port = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_port',
    bucket = gs_bucket,
    source_objects = ['master_data/I94PORT.csv'],
    destination_project_dataset_table = f'{project_id}:{dwh_dataset}.D_PORT',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    skip_leading_rows = 1,
    schema_fields=[
        {'name': 'PORT_ID', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name': 'PORT_NAME', 'type': 'STRING', 'mode': 'NULLABLE'},
    ]
)

load_state = GoogleCloudStorageToBigQueryOperator(
    task_id = 'load_state',
    bucket = gs_bucket,
    source_objects = ['master_data/I94ADDR.csv'],
    destination_project_dataset_table = f'{project_id}:{dwh_dataset}.D_STATE',
    write_disposition='WRITE_TRUNCATE',
    source_format = 'csv',
    skip_leading_rows = 1,
    schema_fields=[
        {'name': 'STATE_ID', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name': 'STATE_NAME', 'type': 'STRING', 'mode': 'NULLABLE'},
    ]
)