Databricks AI Engineer（FDE）完全合格ガイド：Spark、Unity Catalog、RAGから顧客デプロイまで

# RDD (2014) - 低レベルAPI、型安全だが最適化が困難
rdd = sc.parallelize([1, 2, 3, 4, 5])
result = rdd.filter(lambda x: x > 2).map(lambda x: x * 2).collect()

# DataFrame (2015) - SQL親和的、Catalyst Optimizer活用
df = spark.read.parquet("s3://data/events/")
result = df.filter(df.age > 25).groupBy("city").count()

# Dataset (Scalaのみ) - DataFrame + コンパイル時型安全性
# PythonではDataFrameがDataset[Row]に相当

# Catalystの最適化段階を確認する方法
df = spark.read.parquet("s3://data/sales/")
optimized = df.filter("amount > 1000").join(
    spark.read.parquet("s3://data/customers/"),
    "customer_id"
)

# 論理的/物理的実行計画の確認
optimized.explain(True)

# 出力例:
# == Parsed Logical Plan ==
# == Analyzed Logical Plan ==
# == Optimized Logical Plan ==   <-- Predicate Pushdown、Column Pruning適用
# == Physical Plan ==             <-- Join戦略（BroadcastHashJoinなど）決定

# パーティショニング戦略 - データ分布に基づく最適化
# 1. Hash Partitioning（デフォルト）
df.repartition(200, "customer_id")

# 2. Range Partitioning - ソートされたデータに有利
df.repartitionByRange(200, "event_date")

# 3. シャッフル最適化 - broadcast join使用
from pyspark.sql.functions import broadcast

# 小さいテーブル（10MB以下）をブロードキャストしてシャッフルを排除
result = large_df.join(broadcast(small_df), "key")

# 4. Salting - データスキュー解決
from pyspark.sql.functions import concat, lit, rand, floor

# ホットキーにソルトを追加して均等分配
salt_range = 10
skewed_df = skewed_df.withColumn(
    "salted_key",
    concat("key", lit("_"), floor(rand() * salt_range).cast("string"))
)

# AQE (Adaptive Query Execution) - Spark 3.0+
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# PySpark - データサイエンティストとのコラボレーションに有利
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, sum as spark_sum

spark = SparkSession.builder.appName("etl").getOrCreate()

df = (
    spark.read.format("delta")
    .load("s3://lakehouse/bronze/events")
    .filter(col("event_date") >= "2025-01-01")
    .withColumn("category",
        when(col("amount") > 1000, "high")
        .when(col("amount") > 100, "medium")
        .otherwise("low")
    )
    .groupBy("category")
    .agg(spark_sum("amount").alias("total_amount"))
)

// Scala Spark - 型安全性、より良いパフォーマンス
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder.appName("etl").getOrCreate()
import spark.implicits._

val df = spark.read.format("delta")
  .load("s3://lakehouse/bronze/events")
  .filter($"event_date" >= "2025-01-01")
  .withColumn("category",
    when($"amount" > 1000, "high")
    .when($"amount" > 100, "medium")
    .otherwise("low")
  )
  .groupBy("category")
  .agg(sum("amount").alias("total_amount"))

# Delta Lake基本CRUD操作
# 1. テーブル作成
df.write.format("delta").mode("overwrite").saveAsTable("lakehouse.bronze.raw_events")

# 2. 更新（MERGE/UPSERT）
from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "lakehouse.silver.customers")
source = spark.read.format("delta").load("s3://staging/new_customers/")

target.alias("t").merge(
    source.alias("s"),
    "t.customer_id = s.customer_id"
).whenMatchedUpdate(set={
    "name": "s.name",
    "email": "s.email",
    "updated_at": "current_timestamp()"
}).whenNotMatchedInsertAll().execute()

# 3. Time Travel - 過去バージョン照会
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load(path)
df_yesterday = spark.read.format("delta").option("timestampAsOf", "2025-03-22").load(path)

# 4. RESTORE - テーブルを過去バージョンに復元
spark.sql("RESTORE TABLE lakehouse.silver.customers VERSION AS OF 5")

# Bronze Layer - 生データそのまま取り込み（Append-only）
raw_df = (
    spark.readStream
    .format("cloudFiles")  # Auto Loader
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://schema/bronze/events/")
    .load("s3://raw-data/events/")
)

bronze_df = raw_df.withColumn("_ingested_at", current_timestamp())

bronze_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://checkpoints/bronze/events/") \
    .trigger(availableNow=True) \
    .toTable("lakehouse.bronze.raw_events")

# Silver Layer - 精製データ（重複排除、クレンジング、型変換）
silver_df = (
    spark.readStream.table("lakehouse.bronze.raw_events")
    .dropDuplicates(["event_id"])
    .filter(col("event_type").isNotNull())
    .withColumn("amount", col("amount").cast("decimal(18,2)"))
    .withColumn("event_date", to_date(col("event_timestamp")))
)

silver_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://checkpoints/silver/events/") \
    .trigger(availableNow=True) \
    .toTable("lakehouse.silver.clean_events")

# Gold Layer - ビジネス集計テーブル
gold_df = (
    spark.read.table("lakehouse.silver.clean_events")
    .groupBy("customer_id", "event_date")
    .agg(
        count("*").alias("event_count"),
        spark_sum("amount").alias("daily_total"),
        avg("amount").alias("avg_amount")
    )
)

gold_df.write.format("delta").mode("overwrite") \
    .saveAsTable("lakehouse.gold.customer_daily_summary")

-- CDF有効化
ALTER TABLE lakehouse.silver.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

-- CDF照会：変更されたレコードのみ抽出
SELECT * FROM table_changes('lakehouse.silver.customers', 5)
WHERE _change_type IN ('insert', 'update_postimage');

-- Liquid Clustering設定
CREATE TABLE lakehouse.silver.events
CLUSTER BY (event_date, customer_id)
AS SELECT * FROM lakehouse.bronze.raw_events;

-- OPTIMIZEが自動で最適なクラスタリングを適用
OPTIMIZE lakehouse.silver.events;

-- カタログ作成
CREATE CATALOG production;
CREATE CATALOG development;

-- スキーマ（データベース）作成
CREATE SCHEMA production.finance;
CREATE SCHEMA production.marketing;

-- テーブル参照
SELECT * FROM production.finance.transactions
WHERE transaction_date >= '2025-01-01';

-- クロスカタログJoin
SELECT a.*, b.segment
FROM production.finance.transactions a
JOIN production.marketing.customer_segments b
ON a.customer_id = b.customer_id;

# リニエージは自動追跡されるため、追加コード不要
# UIで確認またはREST APIで照会可能

# Unity Catalog REST APIでリニエージ照会
import requests

response = requests.get(
    "https://workspace.cloud.databricks.com/api/2.1/unity-catalog/lineage/table-lineage",
    headers=headers,
    params={"table_name": "production.finance.transactions"}
)

-- Column Level Security：マスキング関数定義
CREATE FUNCTION production.finance.mask_ssn(ssn STRING)
RETURNS STRING
RETURN CASE
    WHEN IS_ACCOUNT_GROUP_MEMBER('finance_admins') THEN ssn
    ELSE CONCAT('***-**-', RIGHT(ssn, 4))
END;

-- カラムにマスキング関数を適用
ALTER TABLE production.finance.customers
ALTER COLUMN ssn SET MASK production.finance.mask_ssn;

-- Row Level Security：行フィルタ関数定義
CREATE FUNCTION production.finance.region_filter(region STRING)
RETURNS BOOLEAN
RETURN CASE
    WHEN IS_ACCOUNT_GROUP_MEMBER('global_admins') THEN TRUE
    WHEN IS_ACCOUNT_GROUP_MEMBER('apac_team') AND region = 'APAC' THEN TRUE
    ELSE FALSE
END;

ALTER TABLE production.finance.transactions
SET ROW FILTER production.finance.region_filter ON (region);

import mlflow
from mlflow.tracking import MlflowClient

# 実験作成・実行
mlflow.set_experiment("/Shared/customer-churn-prediction")

with mlflow.start_run(run_name="xgboost_v2") as run:
    # ハイパーパラメータのロギング
    mlflow.log_param("max_depth", 6)
    mlflow.log_param("learning_rate", 0.1)
    mlflow.log_param("n_estimators", 200)

    # モデル学習
    model = xgb.XGBClassifier(max_depth=6, learning_rate=0.1, n_estimators=200)
    model.fit(X_train, y_train)

    # メトリクスのロギング
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("f1_score", f1_score(y_test, predictions))

    # モデルアーティファクトのロギング
    mlflow.xgboost.log_model(model, "model")

    # Unity Catalogにモデル登録
    mlflow.register_model(
        f"runs:/{run.info.run_id}/model",
        "production.ml_models.churn_predictor"
    )

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# フィーチャーテーブル作成
fe.create_table(
    name="production.features.customer_features",
    primary_keys=["customer_id"],
    timestamp_keys=["event_date"],
    df=customer_features_df,
    description="顧客行動フィーチャー：最近の購入頻度、平均購入金額、最終ログイン日数"
)

# フィーチャールックアップで学習データ生成
from databricks.feature_engineering import FeatureLookup

training_set = fe.create_training_set(
    df=label_df,
    feature_lookups=[
        FeatureLookup(
            table_name="production.features.customer_features",
            lookup_key="customer_id",
            timestamp_lookup_key="event_date"
        )
    ],
    label="churned"
)

training_df = training_set.load_df()

import requests
import json

# Model Serving Endpoint作成（REST API）
endpoint_config = {
    "name": "churn-predictor-endpoint",
    "config": {
        "served_entities": [{
            "entity_name": "production.ml_models.churn_predictor",
            "entity_version": "3",
            "workload_size": "Small",
            "scale_to_zero_enabled": True
        }],
        "auto_capture_config": {
            "catalog_name": "production",
            "schema_name": "ml_monitoring",
            "table_name_prefix": "churn_predictor"
        }
    }
}

# エンドポイントに推論リクエスト送信
response = requests.post(
    "https://workspace.cloud.databricks.com/serving-endpoints/churn-predictor-endpoint/invocations",
    headers=headers,
    json={"dataframe_records": [{"customer_id": "C001", "purchase_count": 12, "avg_amount": 150.0}]}
)

from databricks.vector_search.client import VectorSearchClient

vsc = VectorSearchClient()

# Vector Searchエンドポイント作成
vsc.create_endpoint(name="rag-endpoint", endpoint_type="STANDARD")

# Delta Sync Index作成 - Deltaテーブル変更時に自動同期
vsc.create_delta_sync_index(
    endpoint_name="rag-endpoint",
    index_name="production.rag.document_index",
    source_table_name="production.rag.documents",
    pipeline_type="TRIGGERED",
    primary_key="doc_id",
    embedding_source_column="content",
    embedding_model_endpoint_name="databricks-bge-large-en"
)

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Step 1: Document Chunking
loader = PyPDFLoader("dbfs:/documents/product_manual.pdf")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = text_splitter.split_documents(documents)

# Step 2: Embedding（Databricks Foundation Model使用）
# チャンクをDeltaテーブルに保存するとVector Search Indexが自動エンベディング
chunks_df = spark.createDataFrame([
    {"doc_id": f"doc_{i}", "content": chunk.page_content, "metadata": str(chunk.metadata)}
    for i, chunk in enumerate(chunks)
])
chunks_df.write.format("delta").mode("append").saveAsTable("production.rag.documents")

# Step 3: Search（類似度検索）
results = vsc.get_index("rag-endpoint", "production.rag.document_index").similarity_search(
    query_text="製品の保証ポリシーは何ですか？",
    columns=["content", "metadata"],
    num_results=5
)

# Step 4: Generate（LLMで回答生成）
import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")

context = "\n".join([r["content"] for r in results["result"]["data_array"]])
prompt = f"""以下のコンテキストに基づいて質問に回答してください。

コンテキスト:
{context}

質問: 製品の保証ポリシーは何ですか？
回答:"""

response = client.predict(
    endpoint="databricks-meta-llama-3-1-70b-instruct",
    inputs={"prompt": prompt, "max_tokens": 500, "temperature": 0.1}
)

from databricks.agents import Agent, AgentTool

# エージェントツール定義
search_tool = AgentTool(
    name="document_search",
    description="社内ドキュメントから関連情報を検索します",
    func=lambda query: vsc.get_index("rag-endpoint", "production.rag.document_index")
        .similarity_search(query_text=query, num_results=3)
)

sql_tool = AgentTool(
    name="data_query",
    description="データベースからビジネスデータを照会します",
    func=lambda query: spark.sql(query).toPandas().to_dict()
)

# エージェント作成
agent = Agent(
    model="databricks-meta-llama-3-1-70b-instruct",
    tools=[search_tool, sql_tool],
    system_prompt="あなたはカスタマーサポートエージェントです。ドキュメント検索とデータ照会を通じて正確な回答を提供してください。"
)

# Model Gateway - 複数のLLMプロバイダーを単一インターフェースで統合
# Databricks UIまたはREST APIで設定

# OpenAIモデルをDatabricks Gatewayを通じて呼び出し
response = client.predict(
    endpoint="openai-gpt4-gateway",
    inputs={"messages": [{"role": "user", "content": "分析レポートを作成してください"}]}
)

# Anthropic ClaudeをGatewayを通じて呼び出し
response = client.predict(
    endpoint="anthropic-claude-gateway",
    inputs={"messages": [{"role": "user", "content": "コードをレビューしてください"}]}
)

# オープンソースモデル（Llama、Mistralなど）も同一インターフェース
response = client.predict(
    endpoint="databricks-meta-llama-3-1-70b-instruct",
    inputs={"prompt": "データを要約してください", "max_tokens": 300}
)

# 1. Available Now - 現在利用可能なデータのみ処理して終了（バッチ-ストリーミング統合）
query = (
    df.writeStream
    .format("delta")
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# 2. Processing Time - 指定間隔でマイクロバッチ実行
query = (
    df.writeStream
    .format("delta")
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# 3. Continuous（実験的） - ミリ秒レイテンシのリアルタイム処理
query = (
    df.writeStream
    .format("delta")
    .trigger(continuous="1 second")
    .option("checkpointLocation", checkpoint_path)
    .toTable("lakehouse.silver.events")
)

# イベント時間ベースのウィンドウ集計 + ウォーターマーク
from pyspark.sql.functions import window

windowed_counts = (
    events_df
    .withWatermark("event_time", "10 minutes")  # 10分遅延まで許容
    .groupBy(
        window("event_time", "5 minutes", "1 minute"),  # 5分ウィンドウ、1分スライド
        "device_type"
    )
    .count()
)

windowed_counts.writeStream \
    .format("delta") \
    .outputMode("append") \
    .trigger(processingTime="1 minute") \
    .option("checkpointLocation", checkpoint_path) \
    .toTable("lakehouse.gold.device_counts_5min")

import dlt
from pyspark.sql.functions import col, current_timestamp

# Bronze：生データ収集
@dlt.table(
    name="bronze_events",
    comment="生イベントデータ",
    table_properties={"quality": "bronze"}
)
def bronze_events():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("s3://raw-data/events/")
    )

# Silver：データ品質検証を含む
@dlt.table(
    name="silver_events",
    comment="精製されたイベントデータ"
)
@dlt.expect_or_drop("valid_event_id", "event_id IS NOT NULL")
@dlt.expect_or_fail("valid_amount", "amount >= 0")
@dlt.expect("valid_email", "email RLIKE '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+$'")
def silver_events():
    return (
        dlt.read_stream("bronze_events")
        .dropDuplicates(["event_id"])
        .withColumn("processed_at", current_timestamp())
    )

# Gold：ビジネス集計
@dlt.table(
    name="gold_daily_summary",
    comment="日別ビジネスサマリー"
)
def gold_daily_summary():
    return (
        dlt.read("silver_events")
        .groupBy("event_date", "category")
        .agg(
            count("*").alias("total_events"),
            spark_sum("amount").alias("total_amount")
        )
    )

# Auto Loader - クラウドストレージの新規ファイルを自動検知して取り込み
df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "s3://schema/auto_loader/events/")
    .option("cloudFiles.inferColumnTypes", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("s3://raw-data/events/")
)

# Auto Loaderの利点：
# 1. ファイルリスト管理不要 - 新規ファイル自動検知
# 2. スキーマ推論・進化 - 新カラム自動追加
# 3. Exactly-once処理保証
# 4. 数百万ファイルでも効率的（ファイル通知モード）

# AWSでのDatabricksアーキテクチャ
# 1. ストレージ：S3 + Delta Lake
storage_config = {
    "data_bucket": "s3://company-lakehouse-prod/",
    "checkpoint_bucket": "s3://company-checkpoints-prod/",
    "metastore_bucket": "s3://company-unity-catalog/",
    "encryption": "SSE-KMS",
    "kms_key": "arn:aws:kms:us-east-1:123456789012:key/xxx"
}

# 2. コンピュート：EC2インスタンスベースのクラスター
cluster_config = {
    "spark_version": "14.3.x-scala2.12",
    "node_type_id": "i3.xlarge",
    "driver_node_type_id": "i3.2xlarge",
    "num_workers": 8,
    "autoscale": {"min_workers": 4, "max_workers": 16},
    "aws_attributes": {
        "instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/databricks-role",
        "availability": "SPOT_WITH_FALLBACK",
        "spot_bid_price_percent": 100
    }
}

# AzureでのDatabricksアーキテクチャ
# 1. ストレージ：ADLS Gen2 + Delta Lake
storage_config = {
    "container": "abfss://lakehouse@companystorage.dfs.core.windows.net/",
    "metastore": "abfss://unity-catalog@companystorage.dfs.core.windows.net/",
    "encryption": "Microsoft Managed Keys",
    "network": "Private Endpoint"
}

# 2. ネットワーキング：VNet Injection
# DatabricksワークスペースをカスタマーVNetに直接デプロイ
# - Private Linkでコントロールプレーンアクセス
# - No Public IPでクラスター運用
# - NSG（Network Security Group）でトラフィック制御

# GCPでのDatabricksアーキテクチャ
# 1. ストレージ：GCS + Delta Lake
storage_config = {
    "bucket": "gs://company-lakehouse-prod/",
    "metastore": "gs://company-unity-catalog/",
    "encryption": "Customer-Managed Encryption Key (CMEK)"
}

# 2. BigQuery連携
# DatabricksからBigQueryテーブルを直接照会
bq_df = (
    spark.read
    .format("bigquery")
    .option("table", "project.dataset.table")
    .option("viewsEnabled", "true")
    .load()
)

# TerraformでDatabricksワークスペース + Unity Catalog設定

provider "databricks" {
  alias = "workspace"
  host  = databricks_workspace.main.workspace_url
}

# ワークスペース作成（AWS例）
resource "databricks_workspace" "main" {
  workspace_name = "production-lakehouse"
  region         = "us-east-1"

  aws_config {
    subnet_ids         = var.private_subnet_ids
    security_group_ids = [var.security_group_id]
    instance_profile   = var.instance_profile_arn
  }

  storage_config {
    s3_bucket_name = var.root_bucket_name
  }
}

# Unity Catalog Metastore
resource "databricks_metastore" "main" {
  provider      = databricks.workspace
  name          = "production-metastore"
  storage_root  = "s3://unity-catalog-metastore/"
  force_destroy = false
}

# クラスターポリシー
resource "databricks_cluster_policy" "data_engineering" {
  provider = databricks.workspace
  name     = "Data Engineering Policy"

  definition = jsonencode({
    "spark_version" : { "type" : "fixed", "value" : "14.3.x-scala2.12" },
    "autoscale.max_workers" : { "type" : "range", "maxValue" : 20 },
    "node_type_id" : { "type" : "allowlist", "values" : ["i3.xlarge", "i3.2xlarge"] },
    "custom_tags.team" : { "type" : "fixed", "value" : "data-engineering" }
  })
}