Load data from files to BigQuery using Python

00-bootcamp-project/load_data_to_bigquery.py

@@ -0,0 +1,125 @@
+import json
+
+from google.cloud import bigquery
+from google.oauth2 import service_account
+
+
+DATA_FOLDER = "data"
+
+# ตัวอย่างการกำหนด Path ของ Keyfile ในแบบที่ใช้ Environment Variable มาช่วย
+# จะทำให้เราไม่ต้อง Hardcode Path ของไฟล์ไว้ในโค้ดของเรา
+# keyfile = os.environ.get("KEYFILE_PATH")
+
+keyfile = "YOUR_KEYFILE_PATH"  # แก้ไขชื่อ keyfile ให้ถูกต้อง
+service_account_info = json.load(open(keyfile))
+credentials = service_account.Credentials.from_service_account_info(
+    service_account_info
+)
+project_id = "dataengineercafe"  # แก้ไข project_id ให้สอดคล้องกับ GCP project ของตัวเอง
+client = bigquery.Client(
+    project=project_id,
+    credentials=credentials,
+)
+
+job_config = bigquery.LoadJobConfig(
+    skip_leading_rows=1,
+    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
+    source_format=bigquery.SourceFormat.CSV,
+    autodetect=True,
+)
+
+# Addressess
+data = "addresses"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# Products
+data = "products"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# Order Items
+data = "order_items"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# Promos
+data = "promos"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# ----------
+
+job_config = bigquery.LoadJobConfig(
+    skip_leading_rows=1,
+    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
+    source_format=bigquery.SourceFormat.CSV,
+    autodetect=True,
+    time_partitioning=bigquery.TimePartitioning(
+        type_=bigquery.TimePartitioningType.DAY,
+        field="created_at",
+    ),
+)
+
+# Events
+dt = "2021-02-10"
+partition = dt.replace("-", "")  # 20210210
+data = "events"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}${partition}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# Users
+dt = "2020-10-23"
+partition = dt.replace("-", "")
+data = "users"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}${partition}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")
+
+# Orders
+dt = "2021-02-10"
+partition = dt.replace("-", "")
+data = "orders"
+file_path = f"{DATA_FOLDER}/{data}.csv"
+with open(file_path, "rb") as f:
+    table_id = f"{project_id}.deb_bootcamp.{data}${partition}"
+    job = client.load_table_from_file(f, table_id, job_config=job_config)
+    job.result()
+
+table = client.get_table(table_id)
+print(f"Loaded {table.num_rows} rows and {len(table.schema)} columns to {table_id}")