Skip to content

Commit

Permalink
feat: Onboard World Bank International Debt dataset (#179)
Browse files Browse the repository at this point in the history
  • Loading branch information
dipannitab2392 committed Oct 6, 2021
1 parent 4aba767 commit 5ebbabb
Show file tree
Hide file tree
Showing 19 changed files with 1,439 additions and 0 deletions.
@@ -0,0 +1,42 @@
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# The base image for this build
FROM python:3.8

# Allow statements and log messages to appear in Cloud logs
ENV PYTHONUNBUFFERED True

RUN apt-get -y update && apt-get install -y apt-transport-https ca-certificates gnupg &&\
echo "deb https://packages.cloud.google.com/apt cloud-sdk main" | tee -a /etc/apt/sources.list.d/google-cloud-sdk.list &&\
curl https://packages.cloud.google.com/apt/doc/apt-key.gpg | apt-key add - &&\
apt-get -y update && apt-get install -y google-cloud-sdk

# Copy the requirements file into the image
COPY requirements.txt ./

# Install the packages specified in the requirements file
RUN python3 -m pip install --no-cache-dir -r requirements.txt

# The WORKDIR instruction sets the working directory for any RUN, CMD,
# ENTRYPOINT, COPY and ADD instructions that follow it in the Dockerfile.
# If the WORKDIR doesn’t exist, it will be created even if it’s not used in
# any subsequent Dockerfile instruction
WORKDIR /custom

# Copy the specific data processing script/s in the image under /custom/*
COPY ./csv_transform.py .

# Command to run the data processing script when the container is run
CMD ["python3", "csv_transform.py"]
@@ -0,0 +1,156 @@
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import datetime
import json
import logging
import math
import os
import pathlib
import subprocess
import typing

import pandas as pd
from google.cloud import storage


def main(
source_url: str,
source_file: pathlib.Path,
column_name: str,
target_file: pathlib.Path,
target_gcs_bucket: str,
target_gcs_path: str,
headers: typing.List[str],
rename_mappings: dict,
pipeline_name: str,
) -> None:

logging.info(
f"World Bank Intl Debt {pipeline_name} process started at "
+ str(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
)

logging.info("Creating 'files' folder")
pathlib.Path("./files").mkdir(parents=True, exist_ok=True)

logging.info(f"Downloading file {source_url}")
download_file(source_url, source_file)

logging.info(f"Opening file {source_file}")
df = pd.read_csv(source_file, skip_blank_lines=True)

logging.info(f"Transforming {source_file} ... ")

logging.info(f"Transform: Dropping column {column_name} ...")
delete_column(df, column_name)

logging.info(f"Transform: Renaming columns for {pipeline_name} ...")
rename_headers(df, rename_mappings)

if pipeline_name == "series_times":
logging.info(f"Transform: Extracting year for {pipeline_name} ...")
df["year"] = df["year"].apply(extract_year)
else:
df = df

if pipeline_name == "country_summary":
logging.info("Transform: Creating a new column ...")
df["latest_water_withdrawal_data"] = ""

logging.info("Transform: converting to integer ... ")
df["latest_industrial_data"] = df["latest_industrial_data"].apply(
convert_to_integer_string
)
df["latest_trade_data"] = df["latest_trade_data"].apply(
convert_to_integer_string
)

else:
df = df

logging.info(f"Transform: Reordering headers for {pipeline_name} ...")
df = df[headers]

# print(df.head)

logging.info(f"Saving to output file.. {target_file}")
try:
save_to_new_file(df, file_path=str(target_file))
except Exception as e:
logging.error(f"Error saving output file: {e}.")

logging.info(
f"Uploading output file to.. gs://{target_gcs_bucket}/{target_gcs_path}"
)
upload_file_to_gcs(target_file, target_gcs_bucket, target_gcs_path)

logging.info(
f"World Bank Intl Debt {pipeline_name} process completed at "
+ str(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
)


def download_file(source_url: str, source_file: pathlib.Path) -> None:
subprocess.check_call(["gsutil", "cp", f"{source_url}", f"{source_file}"])


def rename_headers(df: pd.DataFrame, rename_mappings: dict) -> None:
df.rename(columns=rename_mappings, inplace=True)


def delete_column(df: pd.DataFrame, column_name: str) -> None:
df = df.drop(column_name, axis=1, inplace=True)


def extract_year(string_val: str) -> str:
# emaple : YR2018
return string_val[2:]


def save_to_new_file(df: pd.DataFrame, file_path: str) -> None:
df.to_csv(file_path, index=False)


def convert_to_integer_string(input: typing.Union[str, float]) -> str:
str_val = ""
if not input or (math.isnan(input)):
str_val = ""
else:
str_val = str(int(round(input, 0)))
return str_val


def upload_file_to_gcs(file_path: pathlib.Path, gcs_bucket: str, gcs_path: str) -> None:
storage_client = storage.Client()
bucket = storage_client.bucket(gcs_bucket)
blob = bucket.blob(gcs_path)
blob.upload_from_filename(file_path)


if __name__ == "__main__":
logging.getLogger().setLevel(logging.INFO)

main(
source_url=os.environ["SOURCE_URL"],
source_file=pathlib.Path(os.environ["SOURCE_FILE"]).expanduser(),
column_name=os.environ["COLUMN_TO_REMOVE"],
target_file=pathlib.Path(os.environ["TARGET_FILE"]).expanduser(),
target_gcs_bucket=os.environ["TARGET_GCS_BUCKET"],
target_gcs_path=os.environ["TARGET_GCS_PATH"],
headers=json.loads(os.environ["CSV_HEADERS"]),
rename_mappings=json.loads(os.environ["RENAME_MAPPINGS"]),
pipeline_name=os.environ["PIPELINE_NAME"],
)
@@ -0,0 +1,2 @@
pandas
google-cloud-storage
@@ -0,0 +1,39 @@
/**
* Copyright 2021 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


resource "google_bigquery_table" "country_series_definitions" {
project = var.project_id
dataset_id = "world_bank_intl_debt"
table_id = "country_series_definitions"

description = "Country Series Definition table"




depends_on = [
google_bigquery_dataset.world_bank_intl_debt
]
}

output "bigquery_table-country_series_definitions-table_id" {
value = google_bigquery_table.country_series_definitions.table_id
}

output "bigquery_table-country_series_definitions-id" {
value = google_bigquery_table.country_series_definitions.id
}
@@ -0,0 +1,39 @@
/**
* Copyright 2021 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


resource "google_bigquery_table" "country_summary" {
project = var.project_id
dataset_id = "world_bank_intl_debt"
table_id = "country_summary"

description = "Country Summary table"




depends_on = [
google_bigquery_dataset.world_bank_intl_debt
]
}

output "bigquery_table-country_summary-table_id" {
value = google_bigquery_table.country_summary.table_id
}

output "bigquery_table-country_summary-id" {
value = google_bigquery_table.country_summary.id
}
28 changes: 28 additions & 0 deletions datasets/world_bank_intl_debt/_terraform/provider.tf
@@ -0,0 +1,28 @@
/**
* Copyright 2021 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


provider "google" {
project = var.project_id
impersonate_service_account = var.impersonating_acct
region = var.region
}

data "google_client_openid_userinfo" "me" {}

output "impersonating-account" {
value = data.google_client_openid_userinfo.me.email
}
@@ -0,0 +1,39 @@
/**
* Copyright 2021 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


resource "google_bigquery_table" "series_summary" {
project = var.project_id
dataset_id = "world_bank_intl_debt"
table_id = "series_summary"

description = "Series Summary table"




depends_on = [
google_bigquery_dataset.world_bank_intl_debt
]
}

output "bigquery_table-series_summary-table_id" {
value = google_bigquery_table.series_summary.table_id
}

output "bigquery_table-series_summary-id" {
value = google_bigquery_table.series_summary.id
}
39 changes: 39 additions & 0 deletions datasets/world_bank_intl_debt/_terraform/series_times_pipeline.tf
@@ -0,0 +1,39 @@
/**
* Copyright 2021 Google LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/


resource "google_bigquery_table" "series_times" {
project = var.project_id
dataset_id = "world_bank_intl_debt"
table_id = "series_times"

description = "Series Times table"




depends_on = [
google_bigquery_dataset.world_bank_intl_debt
]
}

output "bigquery_table-series_times-table_id" {
value = google_bigquery_table.series_times.table_id
}

output "bigquery_table-series_times-id" {
value = google_bigquery_table.series_times.id
}

0 comments on commit 5ebbabb

Please sign in to comment.