ゼロから実務まで学ぶDatabricks14ステップで、アカウント作成から仕事で使えるところまで

print("はじめてのDatabricks")

# 今どのバージョンのSparkが動いているか
print(spark.version)

%sql
SELECT
    current_catalog()  AS 今のカタログ,   -- データの置き場所（大分類）
    current_schema()   AS 今のスキーマ,   -- その中の仕切り
    current_user()     AS 自分,           -- ログインしている自分
    current_timestamp() AS 今;

%sql
-- ニューヨークのタクシー乗車記録（練習用の定番データ）
SELECT * FROM samples.nyctaxi.trips LIMIT 10;

-- 何行あるか
SELECT count(*) AS 行数 FROM samples.nyctaxi.trips;

%sql
-- どんなテーブルが入っているかを見る
SHOW TABLES IN samples.tpch;

SELECT count(*) AS 注文件数 FROM samples.tpch.orders;

# ざっくりした計算式
料金 = DBU数(1時間あたり) × DBU単価 × 起動していた時間 (+ クラウドのVM代)

# 例：小さめのSQLウェアハウス（サーバーレス, 約4DBU/時）を1時間使うと
4 DBU × $0.70 = 約 $2.8 / 時

# 例：同じものを止め忘れて1日つけっぱなしにすると
4 DBU × $0.70 × 24時間 = 約 $67 / 日

%sql
-- 練習用のカタログ（作れない環境なら、この行は飛ばして既定の workspace を使う）
CREATE CATALOG IF NOT EXISTS learn;

-- 練習用のスキーマ（フォルダ）
CREATE SCHEMA IF NOT EXISTS learn.sales
  COMMENT '学習用。いつ消してもよい';

-- ファイルを置く場所（ボリューム）
CREATE VOLUME IF NOT EXISTS learn.sales.files
  COMMENT '練習用のCSVを置く';

-- これから使う場所を宣言する
USE CATALOG learn;
USE SCHEMA sales;

SELECT current_catalog(), current_schema();

%sql
-- 誰に何を許すか
GRANT SELECT ON TABLE learn.sales.orders TO `analyst@example.com`;

-- 今どんな権限が付いているか
SHOW GRANTS ON TABLE learn.sales.orders;

-- 取り消す
REVOKE SELECT ON TABLE learn.sales.orders FROM `analyst@example.com`;

SELECT   欲しい列
FROM     どのテーブルから
WHERE    どんな条件の行だけ
ORDER BY 何の順に並べる
LIMIT    何件まで

%sql
SELECT
    tpep_pickup_datetime  AS 乗車時刻,
    trip_distance         AS 距離,
    fare_amount           AS 料金,
    pickup_zip            AS 乗車地の郵便番号
FROM samples.nyctaxi.trips
WHERE fare_amount >= 50          -- 料金が50以上の行だけ
ORDER BY fare_amount DESC        -- 料金の大きい順（DESC＝降順）
LIMIT 10;

%sql
SELECT
    trip_distance                          AS 距離,
    fare_amount                            AS 料金,
    round(fare_amount / trip_distance, 1)  AS 単価,          -- 1マイルあたり
    CASE
        WHEN trip_distance < 1  THEN '近距離'
        WHEN trip_distance < 5  THEN '中距離'
        ELSE '長距離'
    END                                    AS 距離区分
FROM samples.nyctaxi.trips
WHERE trip_distance > 0
LIMIT 20;

%sql
SELECT
    pickup_zip           AS 乗車地,
    count(*)             AS 件数,
    round(avg(fare_amount), 1)  AS 平均料金,
    round(sum(fare_amount), 0)  AS 合計料金
FROM samples.nyctaxi.trips
GROUP BY pickup_zip          -- この単位でまとめる
HAVING count(*) >= 100       -- まとめた後の絞り込み
ORDER BY 合計料金 DESC
LIMIT 10;

%sql
SELECT
    date_trunc('MONTH', tpep_pickup_datetime) AS 年月,
    count(*)                                   AS 件数,
    round(sum(fare_amount), 0)                 AS 売上
FROM samples.nyctaxi.trips
GROUP BY 年月           -- ALL と書いてもよい（後述）
ORDER BY 年月;

%sql
-- ① セッションのタイムゾーンを日本にする（そのノートブックの間だけ有効）
SET TIME ZONE 'Asia/Tokyo';

-- ② または、変換関数で明示する（こちらが確実）
SELECT
    tpep_pickup_datetime                              AS utcの時刻,
    from_utc_timestamp(tpep_pickup_datetime, 'Asia/Tokyo') AS 日本時間,
    date_trunc('DAY', from_utc_timestamp(tpep_pickup_datetime, 'Asia/Tokyo')) AS 日本時間の日付
FROM samples.nyctaxi.trips
LIMIT 5;

%sql
SELECT * FROM (
    SELECT
        date_format(o_orderdate, 'yyyy-MM') AS 年月,
        o_orderpriority                      AS 優先度,
        o_totalprice                         AS 金額
    FROM samples.tpch.orders
    WHERE o_orderdate >= DATE'1998-01-01'
)
PIVOT (
    round(sum(金額)/1000, 0)
    FOR 優先度 IN ('1-URGENT', '2-HIGH', '3-MEDIUM')
)
ORDER BY 年月;

%sql
SELECT
    c.c_name        AS 顧客名,
    o.o_orderdate   AS 注文日,
    o.o_totalprice  AS 金額
FROM samples.tpch.orders    AS o          -- 左のテーブル
JOIN samples.tpch.customer  AS c          -- 右のテーブル
  ON o.o_custkey = c.c_custkey            -- つなぐ鍵（キー）
WHERE o.o_orderdate >= DATE'1998-01-01'
ORDER BY o.o_totalprice DESC
LIMIT 10;

%sql
-- 注文が1件も無い顧客も含めて、顧客ごとの注文件数を出す
SELECT
    c.c_name                       AS 顧客名,
    count(o.o_orderkey)            AS 注文件数,     -- ← count(*) にしない
    coalesce(sum(o.o_totalprice), 0) AS 売上
FROM samples.tpch.customer AS c
LEFT JOIN samples.tpch.orders AS o
  ON c.c_custkey = o.o_custkey
GROUP BY ALL
ORDER BY 注文件数 ASC
LIMIT 10;

%sql
-- つなぐ前に「キーが重複していないか」を必ず確認する
SELECT
    count(*)                    AS 行数,
    count(DISTINCT o_orderkey)  AS キーの種類数
FROM samples.tpch.orders;
-- この2つが同じなら、o_orderkey は1行につき1つ（安全に結合できる）

%sql
WITH 対象注文 AS (
    -- ① まず期間で絞る
    SELECT o_custkey, o_orderkey, o_totalprice
    FROM samples.tpch.orders
    WHERE o_orderdate >= DATE'1998-01-01'
),
顧客別 AS (
    -- ② 顧客ごとにまとめる
    SELECT
        o_custkey,
        count(*)          AS 件数,
        sum(o_totalprice) AS 売上
    FROM 対象注文
    GROUP BY ALL
)
-- ③ 最後に顧客名を付ける
SELECT
    c.c_name  AS 顧客名,
    k.件数,
    round(k.売上, 0) AS 売上
FROM 顧客別 AS k
JOIN samples.tpch.customer AS c ON k.o_custkey = c.c_custkey
ORDER BY k.売上 DESC
LIMIT 10;

%sql
USE CATALOG learn;
USE SCHEMA sales;

CREATE TABLE IF NOT EXISTS orders (
    order_id    BIGINT   COMMENT '注文番号',
    customer    STRING   COMMENT '顧客名',
    order_date  DATE     COMMENT '注文日',
    amount      DECIMAL(12,2) COMMENT '金額（円）',
    status      STRING   COMMENT '状態: new / paid / canceled',
    updated_at  TIMESTAMP DEFAULT current_timestamp()
)
COMMENT '学習用の注文テーブル';

DESCRIBE TABLE orders;

%sql
-- 追加
INSERT INTO orders (order_id, customer, order_date, amount, status) VALUES
    (1, '山田商店',   DATE'2026-08-01', 12000, 'paid'),
    (2, '鈴木物産',   DATE'2026-08-02',  8400, 'new'),
    (3, '田中工業',   DATE'2026-08-02', 31500, 'new'),
    (4, '山田商店',   DATE'2026-08-05',  4200, 'canceled');

-- 更新（WHEREを忘れると全行が変わる）
UPDATE orders
SET status = 'paid', updated_at = current_timestamp()
WHERE order_id = 2;

-- 削除
DELETE FROM orders WHERE status = 'canceled';

SELECT * FROM orders ORDER BY order_id;

%sql
-- 顧客ごとの売上サマリを作る
CREATE OR REPLACE TABLE customer_summary AS
SELECT
    customer               AS 顧客,
    count(*)               AS 件数,
    sum(amount)            AS 売上,
    max(order_date)        AS 最終注文日
FROM orders
GROUP BY ALL;

SELECT * FROM customer_summary;

%sql
-- 今日届いた更新データ（ふだんは取り込んだテーブルを使う）
CREATE OR REPLACE TEMP VIEW 本日分 AS
SELECT * FROM VALUES
    (2, '鈴木物産', DATE'2026-08-02',  9000, 'paid'),      -- 金額が変わった
    (5, '佐藤電機', DATE'2026-08-14', 22000, 'new')        -- 新規
AS t(order_id, customer, order_date, amount, status);

MERGE INTO orders AS 元
USING 本日分     AS 新
  ON 元.order_id = 新.order_id              -- 何をもって「同じ行」とするか
WHEN MATCHED THEN UPDATE SET
    元.amount     = 新.amount,
    元.status     = 新.status,
    元.updated_at = current_timestamp()
WHEN NOT MATCHED THEN INSERT
    (order_id, customer, order_date, amount, status)
    VALUES (新.order_id, 新.customer, 新.order_date, 新.amount, 新.status);

SELECT * FROM orders ORDER BY order_id;

%sql
-- 列を足す
ALTER TABLE orders ADD COLUMN memo STRING COMMENT '備考';

-- 列名を変える
ALTER TABLE orders RENAME COLUMN memo TO note;

-- 説明を足す（あとで人が読むときに効く）
COMMENT ON TABLE orders IS '学習用の注文テーブル。毎日MERGEで更新される想定';

-- 中身だけ空にする（テーブルは残る）
-- TRUNCATE TABLE orders;

-- テーブルの詳細（保存場所・サイズ・所有者）
DESCRIBE DETAIL orders;

csv = """order_id,customer,order_date,amount,status
101,山田商店,2026-08-01,12000,paid
102,鈴木物産,2026-08-02,8400,new
103,田中工業,2026-08-02,31500,new
104,,2026-08-03,5000,new
105,佐藤電機,2026-08-04,abc,new
"""

path = "/Volumes/learn/sales/files/orders_20260804.csv"
dbutils.fs.put(path, csv, overwrite=True)

# 置けたか確認
display(dbutils.fs.ls("/Volumes/learn/sales/files/"))

%sql
-- ファイルを直接SELECTする
SELECT * FROM read_files(
    '/Volumes/learn/sales/files/',
    format  => 'csv',
    header  => true,
    inferSchema => true
);

-- 型を推測させず、全部文字として読む（調査のときはこちらが安全）
SELECT * FROM read_files(
    '/Volumes/learn/sales/files/',
    format => 'csv',
    header => true,
    schemaHints => 'order_id STRING, amount STRING'
);

%sql
-- 受け皿（まずは全部STRINGで受ける）
CREATE TABLE IF NOT EXISTS orders_bronze (
    order_id   STRING,
    customer   STRING,
    order_date STRING,
    amount     STRING,
    status     STRING,
    _file_name STRING,
    _loaded_at TIMESTAMP
);

COPY INTO orders_bronze
FROM (
    SELECT
        *,
        _metadata.file_name AS _file_name,   -- どのファイル由来か
        current_timestamp() AS _loaded_at    -- いつ入れたか
    FROM '/Volumes/learn/sales/files/'
)
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'false')
COPY_OPTIONS ('mergeSchema' = 'true');

SELECT * FROM orders_bronze;

%sql
CREATE OR REPLACE TABLE orders_silver AS
SELECT
    cast(order_id AS BIGINT)                       AS order_id,
    nullif(trim(customer), '')                     AS customer,      -- 空文字はNULLに
    try_to_date(order_date, 'yyyy-MM-dd')          AS order_date,
    try_cast(amount AS DECIMAL(12,2))              AS amount,        -- 数値でなければNULL
    lower(trim(status))                            AS status,
    _file_name,
    _loaded_at
FROM orders_bronze
WHERE order_id IS NOT NULL;

-- 品質チェック：変換に失敗した行を数える
SELECT
    count(*)                                          AS 全行,
    count(*) FILTER (WHERE customer IS NULL)          AS 顧客名なし,
    count(*) FILTER (WHERE amount   IS NULL)          AS 金額が変,
    count(*) FILTER (WHERE order_date IS NULL)        AS 日付が変
FROM orders_silver;

%pip install openpyxl
# ライブラリを入れた後はPythonを再起動する
dbutils.library.restartPython()

import pandas as pd

# Excelはpandasで読み、Sparkのデータフレームに変換してテーブルにする
pdf = pd.read_excel("/Volumes/learn/sales/files/売上.xlsx", sheet_name="2026年8月")
df  = spark.createDataFrame(pdf)

df.write.mode("overwrite").saveAsTable("learn.sales.excel_import")

display(spark.table("learn.sales.excel_import"))

from pyspark.sql import functions as F

# テーブルを読む
df = spark.table("samples.nyctaxi.trips")

# 選ぶ・絞る・並べる
result = (
    df
    .select("tpep_pickup_datetime", "trip_distance", "fare_amount")
    .filter(F.col("fare_amount") >= 50)
    .orderBy(F.col("fare_amount").desc())
    .limit(10)
)

display(result)   # 表として画面に出す

from pyspark.sql import functions as F

df = spark.table("samples.nyctaxi.trips")

summary = (
    df
    .withColumn("年月", F.date_trunc("MONTH", "tpep_pickup_datetime"))
    .withColumn("距離区分",
        F.when(F.col("trip_distance") < 1, "近距離")
         .when(F.col("trip_distance") < 5, "中距離")
         .otherwise("長距離"))
    .groupBy("年月", "距離区分")
    .agg(
        F.count("*").alias("件数"),
        F.round(F.avg("fare_amount"), 1).alias("平均料金"),
        F.round(F.sum("fare_amount"), 0).alias("売上"),
    )
    .orderBy("年月", "距離区分")
)

display(summary)

# ① PythonからSQLを実行して、結果をDataFrameで受け取る
df = spark.sql("""
    SELECT pickup_zip, count(*) AS 件数
    FROM samples.nyctaxi.trips
    GROUP BY ALL
""")
display(df)

# ② SQLに値を渡す（文字列を直接つなげない。安全な渡し方）
zip_code = 10001
df2 = spark.sql(
    "SELECT * FROM samples.nyctaxi.trips WHERE pickup_zip = :z LIMIT 5",
    args={"z": zip_code},
)
display(df2)

# ③ DataFrameに名前を付けて、SQLセルから見えるようにする
df.createOrReplaceTempView("zip_summary")

%sql
-- ③で作った一時ビューを、SQLセルからそのまま使える
SELECT * FROM zip_summary ORDER BY 件数 DESC LIMIT 5;

df = spark.table("samples.tpch.orders")          # まだ何も動いていない
big = df.filter(F.col("o_totalprice") > 100000)  # まだ動かない
sel = big.select("o_orderkey", "o_totalprice")   # まだ動かない

sel.count()        # ← ここで初めて計算が走る
display(sel)       # ← ここでも走る

# 同じDataFrameを何度も使うなら、結果を保持させると速い
sel.cache()

# テーブルとして保存（上書き）
summary.write.mode("overwrite").saveAsTable("learn.sales.trip_summary")

# 追記
summary.write.mode("append").saveAsTable("learn.sales.trip_summary")

# 列が増えても受け入れる
(summary.write
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("learn.sales.trip_summary"))

# CSVとして書き出す（人に渡すとき）
(summary.coalesce(1).write
    .mode("overwrite")
    .option("header", "true")
    .csv("/Volumes/learn/sales/files/export/"))

# Sparkで集計してから、小さくなった結果だけpandasに渡す（正しい使い方）
small = spark.sql("SELECT pickup_zip, count(*) AS n FROM samples.nyctaxi.trips GROUP BY ALL")
pdf = small.toPandas()

pdf.plot(kind="bar", x="pickup_zip", y="n", figsize=(10, 4))

%sql
USE CATALOG learn; USE SCHEMA sales;

CREATE OR REPLACE TABLE events (
    event_id BIGINT,
    payload  VARIANT      -- 形が決まっていないデータを入れる型
);

INSERT INTO events VALUES
 (1, parse_json('{"user":{"id":101,"name":"山田"},"action":"purchase","items":[{"sku":"A-1","qty":2},{"sku":"B-7","qty":1}],"amount":12000}')),
 (2, parse_json('{"user":{"id":102,"name":"鈴木"},"action":"view","items":[],"amount":0}')),
 (3, parse_json('{"user":{"id":101,"name":"山田"},"action":"purchase","items":[{"sku":"C-3","qty":5}],"amount":8400}'));

SELECT * FROM events;

%sql
SELECT
    event_id,
    payload:user.id::BIGINT      AS ユーザーid,   -- : で潜り、:: で型を決める
    payload:user.name::STRING    AS 名前,
    payload:action::STRING       AS 行動,
    payload:amount::DECIMAL(12,2) AS 金額,
    payload:items                 AS 明細  -- 配列のまま
FROM events;

%sql
SELECT
    e.event_id,
    e.payload:user.name::STRING AS 名前,
    item:sku::STRING            AS 商品,
    item:qty::INT               AS 数量
FROM events AS e,
     LATERAL variant_explode(e.payload:items) AS t(pos, item)
WHERE e.payload:action::STRING = 'purchase';

%sql
WITH 月別 AS (
    SELECT
        date_format(o_orderdate, 'yyyy-MM') AS 年月,
        sum(o_totalprice)                    AS 売上
    FROM samples.tpch.orders
    WHERE o_orderdate >= DATE'1998-01-01'
    GROUP BY ALL
)
SELECT
    年月,
    round(売上) AS 売上,
    round(lag(売上) OVER (ORDER BY 年月)) AS 前月,
    round(売上 - lag(売上) OVER (ORDER BY 年月)) AS 増減,
    round(sum(売上) OVER (ORDER BY 年月)) AS 累計,
    round(avg(売上) OVER (ORDER BY 年月 ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)) AS 三か月移動平均
FROM 月別
ORDER BY 年月;

%sql
-- 顧客ごとの最新注文だけを残す
SELECT
    o_custkey  AS 顧客,
    o_orderkey AS 注文番号,
    o_orderdate AS 注文日,
    o_totalprice AS 金額
FROM samples.tpch.orders
QUALIFY row_number() OVER (PARTITION BY o_custkey ORDER BY o_orderdate DESC, o_orderkey DESC) = 1
ORDER BY 顧客
LIMIT 10;

%sql
DESCRIBE HISTORY learn.sales.orders;

%sql
-- ① バージョン番号を指定して読む
SELECT * FROM learn.sales.orders VERSION AS OF 1;

-- ② 時刻を指定して読む
SELECT * FROM learn.sales.orders TIMESTAMP AS OF '2026-08-14 10:15:00';

-- ③ 今と過去の差分を見る（何が変わったのか）
SELECT * FROM learn.sales.orders VERSION AS OF 1
EXCEPT
SELECT * FROM learn.sales.orders;

-- ④ 本当に戻す
RESTORE TABLE learn.sales.orders TO VERSION AS OF 1;

%sql
-- 有効にする（作成時に指定してもよい）
ALTER TABLE learn.sales.orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- 変更差分を読む
SELECT * FROM table_changes('learn.sales.orders', 2);
-- _change_type 列に insert / update_preimage / update_postimage / delete が入る

%sql
-- 浅いコピー：実データを複製せず、一瞬で終わる。検証用に最適
CREATE OR REPLACE TABLE learn.sales.orders_test
  SHALLOW CLONE learn.sales.orders;

-- 深いコピー：実データも複製する。バックアップや別環境への移送に
CREATE OR REPLACE TABLE learn.sales.orders_backup
  DEEP CLONE learn.sales.orders;

%sql
-- ① 小さいファイルをまとめる
OPTIMIZE learn.sales.orders;

-- ② よく検索に使う列で、データを整列させておく（推奨のやり方）
ALTER TABLE learn.sales.orders CLUSTER BY (order_date, customer);
OPTIMIZE learn.sales.orders;

-- ③ 統計とファイル数の確認
DESCRIBE DETAIL learn.sales.orders;

%sql
-- 何が消えるかを先に確認する（実際には消さない）
VACUUM learn.sales.orders RETAIN 168 HOURS DRY RUN;

-- 実行（既定は7日=168時間より古い不要ファイルを削除）
VACUUM learn.sales.orders;

%sql
CREATE TABLE IF NOT EXISTS learn.sales.orders_silver (
    order_id   BIGINT,
    customer   STRING,
    order_date DATE,
    amount     DECIMAL(12,2),
    status     STRING,
    _file_name STRING,
    _loaded_at TIMESTAMP
);

MERGE INTO learn.sales.orders_silver AS s
USING (
    SELECT
        cast(order_id AS BIGINT)              AS order_id,
        nullif(trim(customer), '')            AS customer,
        try_to_date(order_date, 'yyyy-MM-dd') AS order_date,
        try_cast(amount AS DECIMAL(12,2))     AS amount,
        lower(trim(status))                   AS status,
        _file_name,
        _loaded_at
    FROM learn.sales.orders_bronze
    WHERE order_id IS NOT NULL
    -- 同じ注文番号が複数あれば、いちばん新しく取り込んだ1件だけ残す
    QUALIFY row_number() OVER (
        PARTITION BY cast(order_id AS BIGINT)
        ORDER BY _loaded_at DESC, _file_name DESC
    ) = 1
) AS b
  ON s.order_id = b.order_id
WHEN MATCHED AND b._loaded_at > s._loaded_at THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

SELECT count(*) AS silver件数 FROM learn.sales.orders_silver;

%sql
CREATE OR REPLACE TABLE learn.sales.monthly_sales_gold AS
SELECT
    date_trunc('MONTH', order_date) AS 年月,
    count(*)                        AS 件数,
    count(DISTINCT customer)        AS 顧客数,
    sum(amount)                     AS 売上,
    round(avg(amount), 0)           AS 平均単価
FROM learn.sales.orders_silver
WHERE status <> 'canceled'
  AND order_date IS NOT NULL
GROUP BY ALL
ORDER BY 年月;

SELECT * FROM learn.sales.monthly_sales_gold;

%sql
-- ① 制約を付けて、違反データが入らないようにする
ALTER TABLE learn.sales.orders_silver
  ADD CONSTRAINT amount_not_negative CHECK (amount IS NULL OR amount >= 0);

-- ② 毎日確認する数字を1本のSQLで出す
SELECT
    current_date()                                          AS 検査日,
    count(*)                                                AS 件数,
    count(*) FILTER (WHERE customer IS NULL)                AS 顧客名なし,
    count(*) FILTER (WHERE amount IS NULL)                  AS 金額なし,
    count(*) - count(DISTINCT order_id)                     AS 重複件数,
    max(_loaded_at)                                         AS 最終取り込み
FROM learn.sales.orders_silver;

# 入力欄を作る（ノートブック上部に表示され、ジョブからも値を渡せる）
dbutils.widgets.text("target_date", "2026-08-14", "処理対象日")

target_date = dbutils.widgets.get("target_date")
print(f"{target_date} 分を処理します")

df = spark.sql(
    "SELECT * FROM learn.sales.orders_silver WHERE order_date = :d",
    args={"d": target_date},
)
display(df)

-- ① 生データを取り込む（ファイルが増えた分だけ自動で処理される）
CREATE OR REFRESH STREAMING TABLE orders_bronze
AS SELECT *, _metadata.file_name AS _file_name, current_timestamp() AS _loaded_at
   FROM STREAM read_files(
        '/Volumes/learn/sales/files/',
        format => 'csv',
        header => true
   );

-- ② 整える。品質ルールを付けられる
CREATE OR REFRESH STREAMING TABLE orders_silver (
    CONSTRAINT 注文番号あり EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT 金額が正    EXPECT (amount >= 0)
)
AS SELECT
    cast(order_id AS BIGINT)          AS order_id,
    nullif(trim(customer), '')        AS customer,
    try_to_date(order_date)           AS order_date,
    try_cast(amount AS DECIMAL(12,2)) AS amount,
    lower(trim(status))               AS status,
    _file_name, _loaded_at
   FROM STREAM(orders_bronze);

-- ③ 集計する
CREATE OR REFRESH MATERIALIZED VIEW monthly_sales_gold
AS SELECT date_trunc('MONTH', order_date) AS 年月,
          count(*) AS 件数, sum(amount) AS 売上
   FROM orders_silver
   WHERE status <> 'canceled'
   GROUP BY ALL;

%sql
-- 3段階すべてに「使ってよい」が必要（ここが最初の関門）
GRANT USE CATALOG ON CATALOG learn            TO `data_analyst`;
GRANT USE SCHEMA  ON SCHEMA  learn.sales      TO `data_analyst`;
GRANT SELECT      ON SCHEMA  learn.sales      TO `data_analyst`;   -- 中の全テーブル

-- 特定のテーブルだけ渡す場合
GRANT SELECT ON TABLE learn.sales.monthly_sales_gold TO `sales_team`;

-- 書き込みも許す（データ担当だけ）
GRANT MODIFY, SELECT ON SCHEMA learn.sales TO `data_engineer`;

-- 確認する
SHOW GRANTS ON SCHEMA learn.sales;
SHOW GRANTS `data_analyst` ON CATALOG learn;

-- 取り消す
REVOKE SELECT ON TABLE learn.sales.orders_bronze FROM `data_analyst`;

%sql
-- ① 列マスク：特定のグループ以外にはぼかして見せる
CREATE OR REPLACE FUNCTION learn.sales.mask_email(email STRING)
RETURN CASE
    WHEN is_account_group_member('pii_reader') THEN email
    ELSE '***@***'
END;

ALTER TABLE learn.sales.customers
  ALTER COLUMN email SET MASK learn.sales.mask_email;

-- ② 行フィルタ：見てよい行だけに絞る
CREATE OR REPLACE FUNCTION learn.sales.area_filter(area STRING)
RETURN is_account_group_member('all_area_reader') OR area = current_user_area();

ALTER TABLE learn.sales.customers
  SET ROW FILTER learn.sales.area_filter ON (area);

%sql
-- 個人情報を含む列に印を付ける（後で一括で探せる）
ALTER TABLE learn.sales.customers
  ALTER COLUMN email SET TAGS ('pii' = 'true');

ALTER SCHEMA learn.sales SET TAGS ('owner_team' = 'データ基盤チーム');

-- 印の付いた列を全社から探す
SELECT * FROM system.information_schema.column_tags
WHERE tag_name = 'pii';

%sql
-- 監査ログ（有効化されている環境で参照できる）
SELECT
    event_time,
    user_identity.email AS 実行者,
    action_name,
    request_params
FROM system.access.audit
WHERE event_date >= current_date() - INTERVAL 7 DAYS
  AND action_name IN ('getTable', 'generateTemporaryTableCredential')
ORDER BY event_time DESC
LIMIT 50;

%sql
-- 直近30日の、種類別・日別のDBU消費
SELECT
    usage_date                                  AS 日付,
    sku_name                                    AS 種類,
    round(sum(usage_quantity), 2)               AS DBU
FROM system.billing.usage
WHERE usage_date >= current_date() - INTERVAL 30 DAYS
GROUP BY ALL
ORDER BY 日付 DESC, DBU DESC;

%sql
-- 何にいくらかかっているか（概算の金額つき）
SELECT
    u.usage_metadata.job_id      AS ジョブid,
    u.sku_name                   AS 種類,
    round(sum(u.usage_quantity), 1)                       AS DBU,
    round(sum(u.usage_quantity * p.pricing.effective_list.default), 2) AS 概算ドル
FROM system.billing.usage AS u
JOIN system.billing.list_prices AS p
  ON u.sku_name = p.sku_name
 AND u.usage_end_time >= p.price_start_time
 AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
WHERE u.usage_date >= current_date() - INTERVAL 30 DAYS
GROUP BY ALL
ORDER BY 概算ドル DESC
LIMIT 20;

%sql
SELECT
    executed_by                                    AS 実行者,
    round(total_duration_ms / 1000, 1)             AS 秒,
    round(read_bytes / 1024 / 1024 / 1024, 2)      AS 読んだGB,
    read_rows                                      AS 読んだ行数,
    left(statement_text, 120)                      AS クエリ
FROM system.query.history
WHERE start_time >= current_date() - INTERVAL 7 DAYS
  AND total_duration_ms > 60000        -- 1分以上かかったもの
ORDER BY total_duration_ms DESC
LIMIT 20;

bundle:
  name: sales_pipeline

resources:
  jobs:
    daily_sales:
      name: 日次売上更新
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "Asia/Tokyo"
      email_notifications:
        on_failure: ["data-team@example.com"]
      tasks:
        - task_key: bronze
          notebook_task:
            notebook_path: ./notebooks/01_bronze.py
        - task_key: silver
          depends_on: [{ task_key: bronze }]
          notebook_task:
            notebook_path: ./notebooks/02_silver.sql
        - task_key: gold
          depends_on: [{ task_key: silver }]
          notebook_task:
            notebook_path: ./notebooks/03_gold.sql

targets:
  dev:
    default: true
    variables: { catalog: learn_dev }
  prod:
    variables: { catalog: learn_prod }

# CLIをインストールして認証（初回のみ）
databricks auth login --host https://xxxx.cloud.databricks.com

# 検証環境へ配る
databricks bundle deploy -t dev

# 本番へ配る
databricks bundle deploy -t prod

# 手動で1回動かす
databricks bundle run daily_sales -t dev

from databricks import sql
import os

with sql.connect(
    server_hostname = os.environ["DATABRICKS_HOST"],
    http_path       = os.environ["DATABRICKS_HTTP_PATH"],   # SQLウェアハウスの接続先
    access_token    = os.environ["DATABRICKS_TOKEN"],       # コードに直接書かない
) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM learn.sales.monthly_sales_gold ORDER BY 年月")
        for row in cur.fetchall():
            print(row)