Databricks で実現するリアルタイム売上集計
Delta Live Tables (DLT) の実践活用ガイド

# 商品マスタのスキーマ
products_schema = StructType([
    StructField("product_id", StringType(), True),  # 商品ID
    StructField("price", IntegerType(), True)       # 単価
])

# 販売トランザクションのスキーマ
transactions_schema = StructType([
    StructField("transaction_id", StringType(), True),  # トランザクションID
    StructField("product_id", StringType(), True),      # 商品ID
    StructField("quantity", IntegerType(), True),       # 売れた数量
    StructField("sale_date", StringType(), True)        # 販売日
])

@dlt.table(
    name="products_raw"
)
def products_raw():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.allowOverwrites", "true")
        .schema(products_schema)
        .load("s3://***/sales/products/")
    )


@dlt.table(
    name="transactions_raw",
    table_properties={"pipelines.reset.allowed": "false"}
)
def transactions_raw():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .schema(transactions_schema)
        .load("s3://***/sales/transactions/")
    )

# 不整合データを検出する関数
def detect_invalid_data(df):
    invalid_df = df.filter(
        col("product_id").isNull() |       # 商品IDがNULLの場合
        col("quantity").isNull() |         # 数量がNULLの場合
        (col("quantity") <= 0) |           # 数量が0以下の場合（括弧追加）
        col("price").isNull() |            # 単価がNULLの場合
        (col("price") <= 0)                # 単価が0以下の場合（括弧追加）
    )
    return invalid_df

# 不整合データを別テーブルに切り出し
@dlt.table(
    name="invalid_transactions"
)
def invalid_transactions():
    products_df = dlt.read("products_raw")
    transactions_df = dlt.read("transactions_raw")

    # 結合後に不整合データを検出
    merged_df = (
        transactions_df.join(products_df, on="product_id", how="inner")
    )
    invalid_df = detect_invalid_data(merged_df)

    return invalid_df

@dlt.table(name="valid_transactions")
def valid_transactions():
    merged_df = dlt.read("transactions_raw").join(dlt.read("products_raw"), "product_id")
    return merged_df.filter(
        col("product_id").isNotNull() &
        col("quantity").isNotNull() & (col("quantity") > 0) &
        col("price").isNotNull()    & (col("price") > 0)
    )

@dlt.table(name="aggregated_sales")
def aggregated_sales():
    return (
        dlt.read("valid_transactions")
        .groupBy("product_id")
        .agg(
            expr("sum(quantity) as total_quantity"),
            expr("sum(quantity * price) as total_revenue")
        )
        .orderBy(col("total_revenue").desc())
    )

select * from catalog00.schema00.aggregated_sales