Databricks AppsとGenieを使ってアンケート集計を行う

はじめに

KCCS デジタルプラットフォーム部の林です。

弊社では定期的に社内で勉強会やハンズオン学習を実施しています。
実施後にアンケートを回答してもらい集計を行うのですが、Genieを使えばもっと簡単に集計や分析ができるのではと考えました。

ついでにDatabricks上でアンケートの回答システムも作成してしまえばデータの収集もできて一石二鳥ではないかと思い、今回はDatabricks上でアンケートシステムの構築からGenieでの分析まで実施してみました。

作成ポイント

今回作成にあたりポイントとした点は以下の通りです。

• アンケート自体はアプリのコードを変更することなく、設定ファイルの定義だけで作成できるようにする
• 事前にテーブルを作成しておかなくてもいいように、アンケート作成に合わせて自動で作成されるようにする

本記事では、Databricks Appsを活用し、JSON設定ファイルから動的にStreamlitフォームを生成し、Delta Tableに直接保存するアンケートシステムを構築します。

システム概要

アーキテクチャ

アンケート設定ファイル → ConfigLoader → FormGenerator
                        　                ↓
     　　Delta Table ← DataManager ← Streamlit UI

アンケート設定ファイル：JSON形式でアンケートの内容を定義する
ConfigLoader：JSONファイルの設定読み込み
FormGenerator：JSON設定情報からStreamlitコンポーネントを動的生成
Streamlit UI：Webでのアンケートページ表示
DataManager：動的テーブル作成、SQLウェアハウス接続
Delta Table：データ保存

アプリの仕様を全て記載すると長くなってしまうため、ポイントに絞って設定内容を解説したいと思います。

1. アンケート設定ファイル

JSON形式でアンケートの内容を定義：

{
  "survey_id": "customer_satisfaction",
  "title": "顧客満足度調査",
  "survey_date": "20241112",
  "questions": [
    {
      "id": "overall_satisfaction",
      "type": "radio",
      "label": "総合的な満足度",
      "required": true,
      "options": ["非常に満足", "満足", "普通", "不満", "非常に不満"]
    },
    {
      "id": "service_rating",
      "type": "slider",
      "label": "サービス評価（1-10点）",
      "required": true,
      "min_value": 1,
      "max_value": 10,
      "default_value": 5
    }
  ]
}

アンケートの内容ではラジオボックスやスライダーなど9種類の質問タイプを利用することができます。
一覧は以下の通りです。

質問タイプ	用途例	Streamlitコンポーネント
text	名前、メールアドレス	st.text_input()
text_area	コメント、意見	st.text_area()
number	年齢、金額	st.number_input()
date	生年月日、予定日	st.date_input()
selectbox	所属部門、職種	st.selectbox()
radio	性別、満足度	st.radio()
checkbox	同意確認、希望有無	st.checkbox()
multiselect	興味分野、改善希望項目	st.multiselect()
slider	評価点数、重要度	st.slider()

2. ConfigLoader

JSON設定の読み込み・検証を行う：

def validate_survey_config(self, config: Dict[str, Any]) -> List[str]:
    """アンケート設定をチェック"""
    errors = []
    
    # 必須フィールドチェック
    required_fields = ['survey_id', 'title', 'survey_date', 'questions']
    for field in required_fields:
        if field not in config:
            errors.append(f"必須フィールド '{field}' が見つかりません")
    
    return errors

3. FormGenerator：

JSON設定から動的にStreamlitコンポーネントを生成する：

def _render_question(self, question: Dict[str, Any]) -> Any:
    """個別の質問を描画"""
    question_type = question['type']
    question_id = question['id']
    label = self._format_label(question)
    help_text = question.get('help_text', None)
        
    # 質問タイプに応じて適切なコンポーネントを選択
    if question_type == 'text':
        return self._render_text_input(question_id, label, question, help_text)
    # 他の質問タイプも同様に実装

動的フォーム生成プロセス：

def render_survey_form(self, survey_config: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """アンケートフォームを描画し、回答データを返す"""
    
    st.title(survey_config['title'])
    if 'description' in survey_config:
        st.markdown(survey_config['description'])
    
    with st.form(key=f"survey_form_{survey_config['survey_id']}"):
        form_data = {}
        
        # 各質問を動的に描画
        for question in survey_config['questions']:
            value = self._render_question(question)
            if value is not None:
                form_data[question['id']] = value
        
        submitted = st.form_submit_button("回答を送信")
        
        if submitted:
            validation_errors = self._validate_form_data(survey_config['questions'], form_data)
            if validation_errors:
                for error in validation_errors:
                    st.error(error)
                return None
            return form_data

4. DataManager

Delta Tableへの操作を担当する：

class DataManager:
    """Delta TableをSQL実行で操作するデータ管理クラス"""
    
    def __init__(self, catalog: str = None, schema: str = None):
        self.catalog = catalog or os.getenv('DEFAULT_CATALOG', 'main')
        self.schema = schema or os.getenv('DEFAULT_SCHEMA', 'survey_data')
        self.warehouse_id = os.getenv('DATABRICKS_WAREHOUSE_ID')
        self._connection = None
        self._ensure_schema_exists()
    
    def _get_connection(self):
        """Databricks SQL接続を取得"""
        if self._connection is None:
            try:
                # Databricks Apps環境では環境変数から自動取得
                self._connection = sql.connect(
                    server_hostname=os.getenv('DATABRICKS_HOST', '').replace('https://', ''),
                    http_path=f"/sql/1.0/warehouses/{self.warehouse_id}",
                    access_token=os.getenv('DATABRICKS_TOKEN')
                )
            except Exception as e:
                st.error(f"Databricks SQL接続に失敗しました: {e}")
                raise
        return self._connection

アンケート設定に基づいて自動的にDelta Tableを作成する。テーブル名はJSON設定から取得した{survey_id}_{survey_date}形式で命名：

def create_table_if_not_exists(self, table_name: str, questions: List[Dict[str, Any]]) -> bool:
    """テーブルが存在しない場合作成"""
    full_table_name = f"`{self.catalog}`.`{self.schema}`.`{table_name}`"
        
    try:
        # テーブル存在チェック
        if self._table_exists(full_table_name):
            return True
            
        # CREATE TABLE SQL文を生成
        columns_sql = []
            
        # システム列を追加
        columns_sql.extend([
            "`response_id` STRING NOT NULL",
            "`submitted_at` TIMESTAMP NOT NULL",
            "`survey_version` STRING"
        ])
            
        # 質問からカラムを生成
        for question in questions:
            field_name = question['id']
            sql_type = self._get_sql_data_type(question)
            nullable = "NOT NULL" if question.get('required', False) else ""
            columns_sql.append(f"`{field_name}` {sql_type} {nullable}".strip())
            
        create_table_sql = f"""
        CREATE TABLE {full_table_name} (
            {', '.join(columns_sql)}
        ) USING DELTA
        TBLPROPERTIES (
            'delta.autoOptimize.optimizeWrite' = 'true',
            'delta.autoOptimize.autoCompact' = 'true'
        )
        """
            
        cursor = self._execute_sql(create_table_sql)
        cursor.close()
            
        st.success(f"テーブル {full_table_name} を作成しました")
        return True
            
    except Exception as e:
        st.error(f"テーブル作成に失敗しました: {e}")
        return False

質問タイプからSQL型を自動判定：

def _get_sql_data_type(self, question: Dict[str, Any]) -> str:
    """質問タイプからSQLデータ型を決定"""
    question_type = question.get('type', 'text')
        
    type_mapping = {
        'text': 'STRING',
        'text_area': 'STRING',
        'number': 'BIGINT' if question.get('min_value', 0) >= 0 else 'DOUBLE',
        'date': 'STRING',  # 日付は文字列として保存
        'selectbox': 'STRING',
        'radio': 'STRING',
        'checkbox': 'BOOLEAN',
        'multiselect': 'STRING',  # カンマ区切り文字列として保存
        'slider': 'BIGINT' if isinstance(question.get('min_value'), int) else 'DOUBLE'
    }
        
    return type_mapping.get(question_type, 'STRING')

実用例：Databricksハンズオン研修アンケート

テストケースとして、Databricksハンズオン研修の受講者アンケートを作成してみます。

プロジェクト構成

databricks-survey-app/
├── app.yml                    # Databricks Apps設定
├── app.py                     # メインアプリケーション
├── requirements.txt           # Python依存関係
├── config/
│   ├── app_config.json       # アプリケーション設定
│   └── surveys/              # アンケート設定ディレクトリ
│       ├── databricks_handson_training.json
│       └── template_survey.json  # アンケートテンプレート（非表示）
├── src/
│   ├── data_manager.py       # Delta Table操作
│   └── form_generator.py     # 動的フォーム生成
├── utils/
│   └── config_loader.py      # 設定ファイル読み込み
└── docs/
    └── setup.md              # セットアップガイド

1.アンケート設計

研修評価アンケート項目：

• 基本情報: 参加者名、所属部署、Databricks / Python経験レベル
• 研修評価: 研修の理解度、難易度、講師評価（スライダー使用）
• 内容分析: 有用だった・興味のあるコンテンツ（複数選択）、改善提案
• 今後の展望: 今後の学習希望分野
• 満足度指標: 全体満足度、他のメンバーへの推奨度

設定例（抜粋）

{
  "survey_id": "databricks_handson_training",
  "title": "Databricksハンズオン研修 受講者アンケート",
  "survey_date": "20241211",
  "questions": [
    {
      "id": "training_content_understanding",
      "type": "slider",
      "label": "研修内容の理解度（1-10）",
      "required": true,
      "min_value": 1,
      "max_value": 10,
      "default_value": 5,
      "help_text": "研修内容をどの程度理解できましたか？"
    },
    {
      "id": "most_valuable_content",
      "type": "multiselect",
      "label": "最も有用だった研修内容",
      "required": false,
      "options": [
        "Databricks基本概念・アーキテクチャ",
        "ノートブックの使い方",
        "Spark DataFrame操作",
        "Delta Lake概要・操作",
        "MLflow機械学習パイプライン"
      ]
    }
  ]
}

生成されるテーブル

CREATE TABLE main.survey_data.databricks_handson_training_20241211 (
  response_id STRING NOT NULL,
  submitted_at TIMESTAMP NOT NULL,
  survey_version STRING,
  participant_name STRING NOT NULL,
  department STRING NOT NULL,
  experience_level STRING NOT NULL,
  training_content_understanding BIGINT NOT NULL,
  most_valuable_content STRING,
  overall_satisfaction BIGINT NOT NULL,
  recommendation_score BIGINT NOT NULL
) USING DELTA

2. Databricks Apps設定

# app.yml
command: ["streamlit", "run", "app.py"]

env:
  - name: STREAMLIT_BROWSER_GATHER_USAGE_STATS
    value: "false"
  - name: STREAMLIT_SERVER_HEADLESS
    value: "true"
  - name: "SURVEY_CONFIG_PATH"
    value: "./config/surveys"
  - name: "APP_CONFIG_PATH"
    value: "./config/app_config.json"
  - name: DEFAULT_CATALOG
    value: "your_catalog_name"
  - name: DEFAULT_SCHEMA
    value: "your_schema_name"
  - name: DATABRICKS_WAREHOUSE_ID
    value: "your-warehouse-id"
  - name: DATABRICKS_TOKEN
    value: "your-access-token"
  - name: DATABRICKS_HOST
    value: "https://your-workspace.cloud.databricks.com/"

3. 必要な権限設定

• Unity Catalogへのアクセス権限
• スキーマ作成・テーブル作成権限
• SQL Warehouseの使用権限

4. アンケート画面

アンケート分析

アンケートの集計ができたらGenieを使って分析を行います。
作成されたテーブルをデータソースとして設定することで、簡単にアンケートの集計を行うことができます。

また複数回実施したデータを登録すれば、回を跨いだ分析も簡単に行うことができます。

まとめ

Databricks AppsとGenieを活用することで簡単にアンケートを集計するシステムが構築できました。
一方、Appsで作成したアプリケーションへの接続にはDatabricksアカウントが必要となるため、アンケートの取得対象者によっては利用するハードルが高いといった課題もあります。
また、アンケートの作成は設定ファイルから簡単にできるものの、JSONファイルでの記載となっているのでこの点もハードルが高い部分であり、AIによる自動作成やGUIから設定可能にするなど、JSONがわからない人でも簡単に利用できるような改善は必要だと感じています。