突然の大量データ分析奮闘記

# monthly_salesジョブのあとにreportジョブが実行されることを保証
monthly_sales
-> report

user_registrations
-> report

▾ project_root/
  ▾ 01_snapshot/
    base.jobnet
    snapshot_sales.ct
    snapshot_sales.job
    snapshot_sales.sql
    ...
  ▾ 02_data_correct/
    base.jobnet
    data_correct_sales.ct
    data_correct_sales.job
    data_correct_sales.sql
    ...
  ▸ 03_calculate/
  ▸ 04_report/
  ▸ 05_check/
  Rakefile

INSERT INTO snapshot_orders
SELECT
  id,
  -- 戻したい時点でキャンセルされていなかったことを表すため、CASE文を使い値を補正
  CASE WHEN cancelled_at < '2017-10-01'
    THEN cancelled_at
    ELSE NULL
  END AS cancelled_at,
  ...
FROM orders
WHERE
  -- レコードの作成日時で絞込み
  created_at < '2017-10-01'

INSERT INTO data_corrected_orders
SELECT
  id,
  -- 実装バグによりcancelled_atが記録されてしまったものを補正する
  CASE
    WHEN id IN (55199, 55201, 55993)
      THEN NULL
    ELSE
      cancelled_at
  END AS cancelled_at,
  ...
FROM
  snapshot_orders
;

INSERT INTO data_corrected_orders
SELECT
  snapshot_orders.id AS id,
  NVL(not_cancelled_orders.cancelled_at, snapshot_orders.cancelled_at) AS cancelled_at,
  ...
FROM
  snapshot_orders,
LEFT JOIN not_cancelled_orders USING (id)

-- 割引適用された売上
INSERT INTO discounted_sales
SELECT
  sales.*
FROM
  sales
JOIN
  discounts ON sales.id=discounts.sales_id

-- 割引適用されていない売上
INSERT INTO not_discounted_sales
SELECT
  sales.*
FROM
  sales
LEFT JOIN
  discounts ON sales.id=discounts.sales_id
WHERE
  discounts.id IS NULL

INSERT INTO monthly_active_user_sales
SELECT
  monthly_active_user_ids.user_id,
  monthly_active_user_ids.month,

  -- LEFT JOINしているため値がNULLになるケースを考慮する
  NVL(monthly_sales.amount, 0) AS sales_amount,
  ...
FROM monthly_active_user_ids
  LEFT JOIN monthly_sales USING(month, user_id)
  LEFT JOIN monthly_fees USING(month, user_id)
  LEFT JOIN monthly_discounts USING(month, user_id)
  ...
;

-- ユーザー別の集計
INSERT INTO user_sales_report
SELECT
  user_id,
  SUM(sales_amount) AS total_sales_amount,
  ...
FROM monthly_active_user_sales
GROUP BY user_id
;

-- 月別の集計
INSERT INTO monthly_sales_report
SELECT
  month,
  SUM(sales_amount) AS sales_amount,
  ...
FROM monthly_active_user_sales
GROUP BY month
;

$ bundle exec bricolage-jobnet --environment=$SALES_SUMMARY_ENV \
                               --variable=closed_on=$SALES_SUMMARY_CLOSED_ON \
                               --variable=schema=$SALES_SUMMARY_SCHEMA \
                               --queue=queue/05_report/jobnet.queue \
                               05_report/base.jobnet

$ bundle exec rake jobnet:report

SALES_SUMMARY_ENV       = ENV['SALES_SUMMARY_ENV'] || 'development'
SALES_SUMMARY_CLOSED_ON = ENV['SALES_SUMMARY_CLOSED_ON']
SALES_SUMMARY_SCHEMA    = ENV['SALES_SUMMARY_SCHEMA']

def jobnet_args(dirname, jobnet_name)
  [
    "--environment=#{SALES_SUMMARY_ENV}",
    "--variable=closed_on=#{SALES_SUMMARY_CLOSED_ON}",
    "--variable=schema=#{SALES_SUMMARY_SCHEMA}",
    "--queue=queue/#{dirname}.#{jobnet_name}.queue",
    "#{dirname}/#{jobnet_name}"
  ]
end

namespace 'jobnet' do
  desc 'Execute all jobs'
  task :all => %i(snapshot data_correct calculate report check)

  desc 'Execute snapshot'
  task :snapshot do
    args = jobnet_args('01_snapshot', 'base.jobnet')
    sh "bundle exec bricolage-jobnet #{args.join(' ')} "
  end

  ...

  desc 'Execute report'
  task :report do
    args = jobnet_args('04_report', 'base.jobnet')
    sh "bundle exec bricolage-jobnet #{args.join(' ')} "
  end

  desc 'Execute check'
  task :check do
    args = jobnet_args('05_check', 'base.jobnet')
    sh "bundle exec bricolage-jobnet #{args.join(' ')} "
  end
end

desc %q(Execute a job file specified by ENV['SALES_SUMMARY_JOB'] )
task :job do
  SALES_SUMMARY_JOB = ENV['SALES_SUMMARY_JOB']

  args = [
    "--environment=#{SALES_SUMMARY_ENV}",
    "--variable=closed_on=#{SALES_SUMMARY_CLOSED_ON}",
    "--variable=schema=#{SALES_SUMMARY_SCHEMA}",
    "--job=#{SALES_SUMMARY_JOB}",
  ]

  sh "bundle exec bricolage #{args.join(' ')}"
end

CREATE TABLE sales_reports (
  month  date  not null,

  "1. 売上額" decimal(19,2) not null,
  "2. 割引額" decimal(19,2) not null,
  ...
)
DISTKEY (month) SORTKEY (month)
;

-- ユーザー別の集計
INSERT INTO user_sales_report
SELECT
  user_id,
  SUM(sales_amount) AS total_sales_amount,
  ...
い
FROM $schema.monthly_active_user_sales -- $schemaを開発者ごとに変更することで同一テーブルを上書きしない
GROUP BY user_id
;

$ bundle exec bricolage-jobnet --variable=schema=xxxx_development ...

SELECT
  user_id,
  '月別売上金額',
  ...
FROM (
  SELECT
    1 AS header,
    'user_id' AS user_id,
    '月別売上金額' AS '月別売上金額',
    ...
  UNION ALL
  SELECT
    0 AS header,
    user_id::text AS user_id,
    monthly_sale_reports.amount::text AS '月別売上金額,
    ...
  FROM monthly_sales_reports
)
ORDER BY header DESC
;

CREATE TABLE sales (
  ...
  amount decimal(19,2) not null encode ZSTD,
  ...
)