当前位置：首页 > news >正文

数据质量转型：自动化 SQL 测试以实现更快速、更智能的分析

news 2026/6/4 5:49:57

原文：towardsdatascience.com/transforming-data-quality-automating-sql-testing-for-faster-smarter-analytics-6da431493570?source=collection_archive---------0-----------------------#2024-10-26

如何测试 SQL 和结果数据集的质量，以回答业务问题并增加客户信任

https://medium.com/@hello.akashm?source=post_page---byline--6da431493570--------------------------------https://towardsdatascience.com/?source=post_page---byline--6da431493570-------------------------------- Akash Mukherjee

·发表于Towards Data Science ·阅读时长 11 分钟·2024 年 10 月 26 日

–

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/4436d79ec654db529fca075e12919af5.png

图片由Caspar Camille Rubin提供，来自Unsplash

在软件开发中，有许多自动化测试工具和框架可以依赖。但是对于分析团队来说，手动测试和数据质量保证（QA）仍然是常态。许多时候，是客户或业务团队首先发现数据质量或完整性的问题，而不是分析团队。

这就是自动化能够产生巨大影响的地方。通过建立一个自动化系统，使用脚本在大规模上运行数据质量测试，你可以保持工作高效而不牺牲数据的准确性或完整性。

当然，当业务问题模糊或开放性时，情况会变得更加复杂。在这种情况下，规则基础逻辑与大语言模型（LLM）的结合可以提供极大帮助——让你生成场景并运行自动化检查。在本教程中，我们将展示如何构建一个自动化测试系统，用来评估和评分数据和 SQL 查询的质量，即使业务问题是以简单的英文表达的。

在我们开始之前，你需要具备的知识

要跟随本教程，请确保你具备以下内容：

对数据库和 SQL 有扎实的理解
有使用 Python 进行 API 调用和数据处理的经验
访问 GPT-4 API 令牌
用于测试的业务问题数据集

设计系统架构

为了构建一个自动化 QA 系统来评估 SQL 查询，架构必须集成基于规则的逻辑、LLM 验证和自动评分。这个设置非常适合处理那些开放式的业务问题，帮助您将测试从手动过程扩展到自动化。

关键组件包括：

查询引擎：接收并执行 SQL 查询的地方。
评估模块：结合静态规则与基于 LLM 的验证结果。
评分系统：根据不同用户角色（如数据科学家、商业领袖和最终用户）对结果进行评分。

该架构包括一个反馈回路，记录问题类型——例如缺失数据、错误粒度或性能缓慢等。这些信息将存储在一个集中式数据库中，以便您可以随时间持续优化系统。我们将使用 Python 进行脚本编写，使用 SQL 跟踪后端问题，使用 OpenAI 的 LLM 解读自然语言输入。通过定期安排这些测试的运行，您将保持一致的数据质量和可扩展性，同时微调查询性能，以与商业目标对齐。

下图展示了数据如何流经系统——从 SQL 摄取到自动化测试、评分和问题跟踪——以便在大规模下保持高数据质量。

最终，这个系统不仅仅是发现错误——它推动持续改进，并使您的技术执行与业务目标保持一致。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/64922a96612de5060deab240c1c68a64.png

作者提供的图像：用来说明技术架构的图示

教程

第 1 步：准备测试问题与答案数据集

要开始，收集内部团队或客户经常向分析团队提出的真实业务问题。许多问题可能是临时的数据请求，因此通过准备多种问题，您可以确保测试具有相关性。以下是一些例子，帮助您入手：

问题 #1：“我们的 Pro Plan 用户中有多少是从试用版转化的？”
问题 #2：“2024 年 6 月我们吸引了多少新用户？”
问题 #3：“目前哪些产品正在流行？”
问题 #4：“我们最畅销的产品目前的销售量是多少？”

第 2 步：构建您的评估与评分标准

2a：定义您的评分员

为了进行彻底的测试，从不同角度设置评分员，以确保覆盖所有方面：

最终用户：关注可用性和清晰度。结果是否易于解读？是否直接解决了原始的商业问题？
数据科学家：评估技术准确性和完整性。所有必要的数据集是否已包含？分析是否详细且可重复？
商业领袖：关注与战略目标的对齐。输出是否支持与业务目标一致的决策？

2b：定义评分标准

每个评分员应根据特定因素来评估查询：

准确性：查询是否提供正确的答案？是否有数据点缺失或误解？
相关性：输出是否包含所有必要的数据，并排除了无关信息？
逻辑：查询是否结构良好？连接、过滤和聚合是否正确应用？
效率：查询是否在没有额外复杂性或延迟的情况下进行了性能优化？

2c：跟踪和记录问题类型

为了涵盖所有方面，记录查询执行过程中常见的问题非常重要。这样可以更容易地标记并运行自动化评估。

错误粒度：数据返回的细节级别不正确。
过多列：结果中包含不必要的字段，造成杂乱。
缺失数据：输出中缺少关键数据。
错误值：计算或数值错误。
性能问题：查询运行效率低，执行时间过长。

importopenaiimportjson# Set your OpenAI API key hereopenai.api_key='your-openai-api-key'defevaluate_sql_query(question,query,results):# Define the prompt with placeholders for question, query, and resultsprompt=f""" As an external observer, evaluate the SQL query and results against the client's question. Provide an assessment from three perspectives: 1\. End User 2\. Data Scientist 3\. Business Leader For each role, provide: 1\. **Overall Score** (0-10) 2\. **Criteria Scores** (0-10): - Accuracy: How well does it meet the question? - Relevance: Is all needed data included, and is irrelevant data excluded? - Logic: Does the query make sense? - Efficiency: Is it concise and free of unnecessary complexity? 3\. **Issue Tags** (2D array: ['tag', 'details']): - Examples: Wrong Granularity, Excessive Columns, Missing Data, Incorrect Values, Wrong Filters, Performance Issues. 4\. **Other Observations** (2D array: ['tag', 'details']) Client Question:{question}SQL Query:{query}SQL Results:{results}Respond ONLY in this format: ```json {{ "endUser": {{"overallScore": "", "criteriaScores": {{"accuracy": "", "relevance": "", "logic": "", "efficiency": ""}}, "issueTags": [], "otherObservations": []}}, "dataScientist": {{"overallScore": "", "criteriaScores": {{"accuracy": "", "relevance": "", "logic": "", "efficiency": ""}}, "issueTags": [], "otherObservations": []}}, "businessLeader": {{"overallScore": "", "criteriaScores": {{"accuracy": "", "relevance": "", "logic": "", "efficiency": ""}}, "issueTags": [], "otherObservations": []}} }} ```py """# Call the OpenAI API with the promptresponse=openai.Completion.create(engine="gpt-4",# or whichever model you're usingprompt=prompt,max_tokens=500,# Adjust token size based on expected response lengthtemperature=0# Set temperature to 0 for more deterministic results)# Parse and return the resultreturnjson.loads(response['choices'][0]['text'])# Example usagequestion="How many Pro Plan users converted from trial?"query="SELECT COUNT(*) FROM users WHERE plan = 'Pro' AND status = 'Converted' AND source = 'Trial';"results="250"evaluation=evaluate_sql_query(question,query,results)print(json.dumps(evaluation,indent=4))

forquestion,query,resultsintest_cases:# Call the OpenAI API to evaluate the SQL query and resultsresponse=openai.Completion.create(engine="text-davinci-003",# Replace with GPT-4 or relevant engineprompt=prompt.format(question=question,query=query,results=results),max_tokens=1000)# Process and store the responseprocess_response(response)defstore_results_in_db(test_run_id,question,role,scores,issue_tags,observations):# SQL insert query to store evaluation results in the issue cataloginsert_query=""" INSERT INTO issue_catalog (test_run_id, question, role, overall_score, accuracy_score, relevance_score, logic_score, efficiency_score, issue_tags, other_observations) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s); """db_cursor.execute(insert_query,(test_run_id,question,role,scores['overall'],scores['accuracy'],scores['relevance'],scores['logic'],scores['efficiency'],json.dumps(issue_tags),json.dumps(observations)))db_conn.commit()

设置问题目录表

issue_catalog表作为存储详细测试结果的主要库，帮助你清晰地跟踪查询性能并标记长期存在的问题。通过使用 JSONB 格式存储问题标签和观察结果，你可以获得更大的灵活性，允许你记录复杂信息，而无需频繁更新数据库架构。以下是设置的 SQL 代码：

CREATE TABLE issue_catalog(idSERIAL PRIMARY KEY,test_run_id INT NOT NULL,question TEXT NOT NULL,role TEXT NOT NULL,--e.g.,endUser,dataScientist,businessLeader overall_score INT NOT NULL,accuracy_score INT NOT NULL,relevance_score INT NOT NULL,logic_score INT NOT NULL,efficiency_score INT NOT NULL,issue_tags JSONB,--Storing issue tagsasJSONforflexibility other_observations JSONB,--Storing other observationsasJSON created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);