多表格關聯資料

import pandas as pd

# 讀取原始資料表
companies = pd.read_csv('companies.csv')
applications = pd.read_csv('applications.csv')
tracking = pd.read_csv('tracking.csv')

# 標記每個公司的第一次和最新一次申請
applications['sort_tuple'] = list(zip(applications['apply_date'], applications['application_id']))

# 找出每個公司的最早申請
min_tuples = applications.groupby('company_id')['sort_tuple'].transform('min')
applications['is_first_application'] = (applications['sort_tuple'] == min_tuples)

# 找出每個公司的最晚申請
max_tuples = applications.groupby('company_id')['sort_tuple'].transform('max')
applications['is_latest_application'] = (applications['sort_tuple'] == max_tuples)

applications.drop(columns=['sort_tuple'], inplace=True, errors='ignore')

# 將財務追蹤資料串接上申請資料，以獲得公司編號
tracking_w_company = tracking.merge(
    applications[['company_id', 'application_id']],
    how='left',
    left_on='application_id',
    right_on='application_id'
)

# 標記每個公司的最新一次財務追蹤
tracking_w_company['sort_tuple'] = list(zip(
    tracking_w_company['tracking_date_last_tracking_date'],
    tracking_w_company['application_id']
))

max_tuples = tracking_w_company.groupby('company_id')['sort_tuple'].transform('max')
tracking_w_company['is_latest_tracking'] = (tracking_w_company['sort_tuple'] == max_tuples)

tracking_w_company.drop(columns=['sort_tuple'], inplace=True, errors='ignore')

# 合併企業資料與申請資料
denorm_data = companies.merge(
    applications[applications['is_first_application']].add_prefix('first_apply_'),
    how='left',
    left_on='company_id',
    right_on='first_apply_company_id'
).drop(columns=['first_apply_company_id', 'first_apply_is_first_application', 'first_apply_is_latest_application']).merge(
    applications[applications['is_latest_application']].add_prefix('latest_apply_'),
    how='left',
    left_on='company_id',
    right_on='latest_apply_company_id'
).drop(columns=['latest_apply_company_id', 'latest_apply_is_first_application', 'latest_apply_is_latest_application'])

# 加入彙整後的追蹤資料
denorm_data = denorm_data.merge(
    tracking_w_company[tracking_w_company['is_latest_tracking']].drop(columns=['sort_tuple'], errors='ignore').add_prefix('latest_track_'),
    how='left',
    left_on='company_id',
    right_on='latest_track_company_id'
).drop(columns=['latest_track_company_id', 'latest_track_is_latest_tracking'])

# 儲存整合後的寬表
denorm_data.to_csv('denormalized_data.csv', index=False)

-- 在資料庫中直接執行反正規化
WITH first_applications AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY company_id ORDER BY apply_date) as rn
    FROM applications
),
latest_applications AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY company_id ORDER BY apply_date DESC) as rn
    FROM applications
),
latest_tracking AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY application_id ORDER BY tracking_date DESC) as rn
    FROM tracking
)
SELECT
    c.*,
    fa.* AS first_apply_,
    la.* AS latest_apply_,
    lt.* AS latest_track_
FROM companies c
LEFT JOIN first_applications fa ON c.company_id = fa.company_id AND fa.rn = 1
LEFT JOIN latest_applications la ON c.company_id = la.company_id AND la.rn = 1
LEFT JOIN latest_tracking lt ON la.application_id = lt.application_id AND lt.rn = 1;

Loader:
  data:
    filepath: 'denormalized_data.csv'
    schema: 'denormalized_schema.yaml'
Preprocessor:
  default:
    method: 'default'
Synthesizer:
  default:
    method: 'default'
Postprocessor:
  default:
    method: 'default'
Reporter:
  output:
    method: 'save_data'
    source: 'Postprocessor'