如何从Python Dash应用发送POST请求将数据保存至Azure Databricks表

我用Python Dash构建了一个Web应用，已经在Azure Databricks中配置好数据库和表。现在需要将UI中收集的数据写入Databricks的对应表中，目前已能成功建立连接并执行查询操作，但无法实现数据写入，请问该如何解决？

现有代码 #

databricks_connection.py #

import pandas as pd
import os
from pyspark.sql import SparkSession

class DatabricksConnection:
    def __init__(self):
        self.server = os.environ.get("SERVER_NAME")
        self.http_path = os.environ.get("HTTP_PATH")
        self.client_id = os.environ.get("CLIENT_ID")
        self.client_secret = os.environ.get("CLIENT_SECRET")
        self.connection = None

    def connect(self):
        self.spark = SparkSession.builder.getOrCreate()

    def query(self, query_string, chunk_size=None):
        if not hasattr(self, 'spark'):
            from pyspark.sql import SparkSession
            self.spark = SparkSession.builder.getOrCreate()
        results = self.spark.sql(query_string)
        results = results.toPandas()
        return results

data.py #

import sys
import os
from pathlib import Path
from databricks_connection import DatabricksConnection as conn

class data:
    def __init__(self):
        self.Databricks = conn.DatabricksConnection()
        self.Databricks.connect()
        # 补充：初始化查询文件目录路径
        self.queries_dir = os.path.join(os.path.dirname(__file__), "queries")

    def _load_query(self, query_file):
        with open(query_file, 'r') as file:
            return file.read()

    def _format_query(self, query_template, **kwargs):
        return query_template.format(**kwargs)

    def fetch_data(self, query_file, execute = True, **kwargs):
        query_template = self._load_query(os.path.join(self.queries_dir, query_file))
        query_string = self._format_query(query_template, **kwargs)
        self.last_query_string = query_string
        if execute:
            return self.Databricks.query(query_string)
        else:
            return query_string

data = data()        
def test(parameter):
    return data.fetch_data('query.sql', parameter=parameter)

1. 完善Databricks连接配置 #

当前connect方法未正确配置Azure Databricks的认证参数，需修改为指向云端集群的配置：

def connect(self):
    self.spark = SparkSession.builder \
        .config("spark.databricks.service.server.enabled", "true") \
        .config("spark.databricks.service.server.hostname", self.server) \
        .config("spark.databricks.service.server.http.path", self.http_path) \
        .config("spark.databricks.service.client.id", self.client_id) \
        .config("spark.databricks.service.client.secret", self.client_secret) \
        .getOrCreate()

2. 为连接类添加数据写入方法 #

在DatabricksConnection类中添加两种写入方式，按需选择：

方式一：批量写入DataFrame到表

适合一次性写入多条数据：

def save_data(self, df, table_name, mode="append"):
    # 转换Pandas DataFrame为Spark DataFrame（如果传入的是Pandas对象）
    if isinstance(df, pd.DataFrame):
        spark_df = self.spark.createDataFrame(df)
    else:
        spark_df = df
    # mode参数支持：append（追加）、overwrite（覆盖）、ignore（忽略重复）
    spark_df.write.mode(mode).saveAsTable(table_name)

方式二：执行自定义SQL写入语句

适合需要复杂逻辑的INSERT/UPDATE操作：

def execute_write_query(self, query_string):
    # 执行非查询类SQL语句
    self.spark.sql(query_string)

3. 在data类中封装写入接口 #

对应DataFrame写入方式：

def save_to_table(self, df, table_name, mode="append"):
    self.Databricks.save_data(df, table_name, mode)

对应SQL写入方式：

def run_write_query(self, query_file, **kwargs):
    query_template = self._load_query(os.path.join(self.queries_dir, query_file))
    query_string = self._format_query(query_template, **kwargs)
    self.Databricks.execute_write_query(query_string)

4. Dash回调中调用写入逻辑 #

将UI收集的数据整理后调用保存方法，示例如下：

# Dash回调示例
@app.callback(
    Output('submit-status', 'children'),
    Input('submit-button', 'n_clicks'),
    State('input-field-1', 'value'),
    State('input-field-2', 'value'),
    prevent_initial_call=True
)
def submit_data(n_clicks, val1, val2):
    # 将UI数据转换为DataFrame
    data_df = pd.DataFrame({
        "column1": [val1],
        "column2": [val2]
    })
    # 写入Databricks指定表
    data.save_to_table(data_df, "database_name.table_name", mode="append")
    return "数据保存成功"

5. 权限验证 #

确保client_id对应的服务主体在Azure Databricks中拥有目标表的写入权限（如INSERT、UPDATE权限），可在Databricks UI的表权限设置中配置。

内容的提问来源于stack exchange，提问作者Jason Susanto