หมายเหตุ
การเข้าถึงหน้านี้ต้องได้รับการอนุญาต คุณสามารถลอง ลงชื่อเข้าใช้หรือเปลี่ยนไดเรกทอรีได้
การเข้าถึงหน้านี้ต้องได้รับการอนุญาต คุณสามารถลองเปลี่ยนไดเรกทอรีได้
ในสถาปัตยกรรมเหรียญ คุณต้องบังคับใช้คุณภาพของข้อมูลในทุกขั้นตอน คุณภาพข้อมูลที่ไม่ดีอาจนําไปสู่ข้อมูลเชิงลึกที่ไม่ถูกต้องและความไร้ประสิทธิภาพในการปฏิบัติงาน
บทความนี้อธิบายวิธีการใช้การตรวจสอบคุณภาพข้อมูลในมุมมองทะเลสาบที่เป็นรูปธรรม (MLV) ใน Microsoft Fabric
นําคุณภาพข้อมูลไปใช้
ข้อจํากัดระดับแถวจะตรวจสอบแต่ละแถวแยกกัน (เช่น "customerName ไม่ใช่ null ในแถวนี้") อย่างไรก็ตาม ปัญหาคุณภาพที่แท้จริงหลายอย่างเป็น คุณสมบัติของคอลัมน์โดยรวม:
-
ความเป็นเอกลักษณ์ — คีย์หลักหรือคีย์ธุรกิจ เช่น
order_idไม่มีคีย์ซ้ํา -
ความสมบูรณ์ — คอลัมน์สําคัญที่ไม่มี
customer_idค่าว่าง - รูปร่างทางสถิติ — ค่าเฉลี่ย ขนาด หรือการแจกแจงของคอลัมน์จะอยู่ในช่วงที่คาดหวัง
การจับปัญหาเหล่านี้ในจุดรีเฟรชจะหยุดข้อมูลที่ไม่ดีไม่ให้แพร่กระจายลงไปในสายโซ่ทองแดง→เงิน→ทองของคุณ และให้สัญญาณคุณภาพที่ตรวจสอบได้ในทุกครั้งที่วิ่ง
ข้อจํากัดคุณภาพข้อมูลระดับแถว
ในมุมมองทะเลสาบที่สร้างขึ้นจริง (MLVs) ใน Fabric ให้รักษาคุณภาพข้อมูลโดยการกําหนดข้อจํากัดในมุมมองของคุณ หากไม่มีการตรวจสอบที่ชัดเจนปัญหาข้อมูลเล็กน้อยอาจเพิ่มเวลาในการประมวลผลหรือทําให้ไปป์ไลน์ล้มเหลว
เมื่อแถวละเมิดข้อจํากัด คุณสามารถใช้การดําเนินการอย่างใดอย่างหนึ่งต่อไปนี้:
ล้มเหลว: หยุดการรีเฟรช MLV เมื่อมีการละเมิดข้อจํากัดครั้งแรก นี่คือลักษณะการทํางานเริ่มต้น แม้ว่าคุณจะไม่ได้ระบุ
FAILDROP: ประมวลผลต่อไปและลบเรกคอร์ดที่ละเมิดข้อจํากัด มุมมองสายข้อมูลแสดงจํานวนเรกคอร์ดที่ลดลง
หมายเหตุ
ถ้าคุณกําหนดทั้งการดําเนินการ DROP และ FAIL ใน MLV การดําเนินการ FAIL จะมีความสําคัญเหนือกว่า
กําหนดการตรวจสอบคุณภาพข้อมูลในมุมมองทะเลสาบที่เป็นรูปธรรม
เมื่อคุณสร้างมุมมองทะเลสาบที่เป็นรูปธรรม คุณสามารถกําหนดข้อจํากัด ซึ่งเป็นกฎคุณภาพข้อมูลที่ตรวจสอบความถูกต้องของแต่ละแถวในระหว่างการรีเฟรช ข้อจํากัดคือนิพจน์บูลีนที่ทุกแถวต้องตอบสนอง แถวที่ผ่านจะถูกเขียนลงในตารางผลลัพธ์ แถวที่ล้มเหลวจะได้รับการจัดการตามการตั้งค่าการละเมิด: แถวเหล่านี้จะถูกทิ้งอย่างเงียบ ๆ หรือทําให้การรีเฟรชทั้งหมดล้มเหลว
ตัวอย่างต่อไปนี้กําหนดข้อจํากัด cust_blankซึ่งจะตรวจสอบว่า customerName เขตข้อมูลไม่ใช่ null หรือไม่ ข้อจํากัดไม่รวมแถวที่มีค่า Null customerName จากการประมวลผล
CREATE OR REPLACE MATERIALIZED LAKE VIEW IF NOT EXISTS silver.customers_enriched
(CONSTRAINT cust_blank CHECK (customerName is not null) on MISMATCH DROP)
AS
SELECT
c.customerID,
c.customerName,
c.contact,
CASE
WHEN COUNT(o.orderID) OVER (PARTITION BY c.customerID) > 0 THEN TRUE
ELSE FALSE
END AS has_orders
FROM bronze.customers c LEFT JOIN bronze.orders o
ON c.customerID = o.customerID;
ฟังก์ชันในตัวระบบ
ฟังก์ชัน Spark/SQL ในตัว เช่น UPPER(), LOWER(), TRIM(), COALESCE(), INITCAP() และ DATE_FORMAT() ได้รับการสนับสนุนอย่างเต็มที่ในบริบท MLV ทั้งหมดสําหรับทั้ง CREATE และ LINEAGE รีเฟรช
CREATE MATERIALIZED LAKE VIEW sample_lakehouse.silver.names (
CONSTRAINT substring_check
CHECK (SUBSTRING(name, 1, 2) = 'Al') ON MISMATCH drop
) AS
SELECT id, name
FROM (VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Ann')) AS t(id, name)
หมายเหตุ
ฟังก์ชันของระบบเป็นตัวเลือกที่ง่ายและน่าเชื่อถือที่สุด พวกเขาไม่จําเป็นต้องลงทะเบียน ทํางานในทุกบริบท และได้รับการสนับสนุนอย่างเต็มที่ในระหว่างการรีเฟรช LINEAGE
UDFs – กําหนดและลงทะเบียนในสมุดบันทึกเดียวกัน
UDF ที่ลงทะเบียนด้วย spark.udf.register() ในสมุดบันทึกเดียวกันได้รับการสนับสนุนสําหรับ CREATE ในทุกบริบท สําหรับการรีเฟรช LINEAGE เฉพาะบริบท PySpark เท่านั้นที่ได้รับการสนับสนุน เนื่องจากข้อกําหนด UDF ทํางานเป็นส่วนหนึ่งของการดําเนินการสมุดบันทึกตามกําหนดการ
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import expr
spark = SparkSession.builder.getOrCreate()
# UDF: Extract domain from email
def extract_email_domain(email):
if email is None or '@' not in email:
return None
return email.split('@')[1]
# Registration
spark.udf.register(
"udf_email_domain",
extract_email_domain,
StringType()
)
@fmlv.materialized_lake_view(
name="udf_testing_silver.mlv_high_value_customers",
comment="High-value customers identified by UDF criteria",
table_properties={"delta.enableChangeDataFeed": "true"}
)
def mlv_high_value_customers():
return spark.sql("""
SELECT
c.customer_id,
c.name,
c.email,
udf_email_domain(c.email) as email_domain,
c.segment,
c.lifetime_value,
total_transactions.total_amount,
total_transactions.txn_count
FROM udf_testing_bronze.customers c
INNER JOIN (
SELECT
customer_id,
SUM(amount) as total_amount,
COUNT(*) as txn_count
FROM udf_testing_bronze.transactions
WHERE udf_is_positive(amount)
GROUP BY customer_id
HAVING SUM(amount) > 1000
) total_transactions ON c.customer_id = total_transactions.customer_id
WHERE udf_validate_customer(c.email, c.age)
AND c.segment IN ('premium', 'vip')
""")
print("✓ Created mlv_high_value_customers")
ไลบรารีของบุคคลที่สาม - Pandas UDFs
ไลบรารีของบุคคลที่สาม เช่น Pandas UDF อนุญาตให้ใช้กฎคุณภาพของข้อมูลด้วยการประมวลผลแบบเวกเตอร์ เปิดใช้งานการตรวจสอบขั้นสูง เช่น ตรรกะทางธุรกิจแบบกําหนดเอง การตรวจสอบทางสถิติ หรือการตรวจหารูปแบบที่ไม่สามารถทําได้ด้วยฟังก์ชันในตัว สิ่งนี้ช่วยสร้างข้อจํากัดด้านคุณภาพข้อมูลที่ปรับขนาดได้และนํากลับมาใช้ใหม่ได้ในระหว่างการสร้างและรีเฟรช MLV
import fmlv
from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
from datetime import datetime
import pandas as pd
from pyspark.sql.types import BooleanType
def pandas_check_impl(val):
# Reject if value < median of [100, 200, 300]
return val >= pd.Series([100, 200, 300]).median()
spark.udf.register("pandas_check", pandas_check_impl, BooleanType())
@fmlv.materialized_lake_view(
name="silver.pyspark_from_two_sqlmlv_inner_pandas",
comment="PySpark MLV INNER JOIN using pandas-based constraint and DROP violations"
)
@fmlv.check(
name="dq_pandas_check",
condition="pandas_check(l3)",
action="DROP"
)
def pyspark_from_two_sqlmlv_inner_pandas():
# Define the function
# Register the function as a Spark UDF
# Read source tables
df1 = spark.table("silver.base_sqlmlv")
df2 = spark.table("silver.base_sqlmlv")
# Rename columns for unique join
df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])
# Perform INNER JOIN
df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
return df
df = spark.table("silver.pyspark_from_two_sqlmlv_inner_pandas")
# All amounts should be >= median (200)
assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
print("PySpark MLV INNER JOIN with pandas-based DQ DROP passed")
ไลบรารีที่กําหนดเอง – วงล้อ Python (.whl)
ฟังก์ชันที่บรรจุเป็นไฟล์ jar หรือไฟล์ล้อสามารถติดตั้งบนคลัสเตอร์ Fabric (ผ่านการตั้งค่าสภาพแวดล้อม) และใช้ในข้อกําหนด MLV CREATE และ LINEAGE REFRESH ได้รับการสนับสนุนสําหรับบริบท PySpark ดู จัดการไลบรารีแบบกําหนดเองในสภาพแวดล้อม Fabric สําหรับรายละเอียดเพิ่มเติม
%%pyspark
import fmlv
from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
from datetime import datetime
from pyspark.sql.types import BooleanType
from custom_dq_lib import threshold_check
def custom_check_impl(val):
return threshold_check(val, threshold=200)
spark.udf.register("custom_check", custom_check_impl, BooleanType())
@fmlv.materialized_lake_view(
name="silver.pyspark_from_two_sqlmlv_inner_custom_whl",
comment="PySpark MLV INNER JOIN using custom DQ library and DROP violations",
replace=True
)
@fmlv.check(
name="dq_custom_check",
condition="custom_check(l3)",
action="DROP"
)
def pyspark_from_two_sqlmlv_inner_custom():
# Wrap the custom function as Spark UDF
# Read source tables
df1 = spark.table("silver.base_sqlmlv")
df2 = spark.table("silver.base_sqlmlv")
# Rename columns for unique join
df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])
# Perform INNER JOIN
df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
return df
df = spark.table("silver.pyspark_from_two_sqlmlv_inner_custom_whl")
# All amounts should be >= threshold (200)
assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
print("PySpark MLV INNER JOIN with custom DQ library passed")
ฟังก์ชันข้อมูลผู้ใช้ Fabric
ฟังก์ชันข้อมูลผู้ใช้ Fabric (UDF) ถูกกําหนดและจัดการจากส่วนกลางในพื้นที่ทํางาน Fabric พร้อมใช้งานสําหรับโน้ตบุ๊กหรือไปป์ไลน์ใดๆ โดยไม่จําเป็นต้องลงทะเบียนใหม่ต่อเซสชัน ทําให้เหมาะสําหรับไปป์ไลน์ MLV ที่ใช้งานจริง คุณลักษณะนี้ได้รับการสนับสนุนเฉพาะในบริบท PySpark สําหรับทั้ง CREATE และการรีเฟรช LINEAGE เรียนรู้เพิ่มเติมเกี่ยวกับฟังก์ชันข้อมูลผู้ใช้ที่นี่ สําหรับข้อมูลเพิ่มเติม โปรดดู ภาพรวมฟังก์ชันข้อมูลผู้ใช้ Fabric
%%pyspark
import fmlv
from pyspark.sql import functions as F
from notebookutils import udf
myFuncs = udf.getFunctions("UserDataFunction_1")
def add_greeting_column(df):
pdf = df.toPandas()
pdf["greeting"] = pdf["name"].apply(lambda n: myFuncs.hello_fabric(n))
return spark.createDataFrame(pdf)
import fmlv
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType
# -------------------------------------------------------------
# Define base PySpark MLV (no SQL)
# -------------------------------------------------------------
@fmlv.materialized_lake_view(
name="silver.base_pysparkmlv",
comment="Base MLV created using PySpark",
replace=True
)
def base_pysparkmlv():
schema = StructType([
StructField("id", IntegerType(), True),
StructField("name", StringType(), True),
StructField("amount", DoubleType(), True),
StructField("country", StringType(), True)
])
data = [
(1, "Alice", 100.0, "US"),
(2, "Bob", 200.0, "UK"),
(3, "Charlie", 300.0, "UK")
]
return spark.createDataFrame(data, schema)
@fmlv.materialized_lake_view(
name="silver.mlv_udfn_null_test",
replace=False
)
@fmlv.check(
name="null_check",
condition="greeting IS NOT NULL",
action="FAIL"
)
def mlv_udfn_null_test():
df = spark.table("silver.base_pysparkmlv")
return add_greeting_column(df)
การตรวจสอบคุณภาพข้อมูลในระดับคอลัมน์
วิธีการทำงาน
กําหนดการตรวจสอบคุณภาพข้อมูลระดับคอลัมน์ (DQ) ด้วยตัวตกแต่ง Python สามตัวที่คุณวางไว้ในเซลล์สมุดบันทึก แยกจากการกําหนด MLV:
| ตกแต่ง | Purpose |
|---|---|
@fmlv.dq_suite(mlv_name=…) |
ลงทะเบียนชุดคุณภาพข้อมูลกับ MLV ที่มีอยู่แล้วและโหลดข้อมูลเพื่อใช้งานตรวจสอบ |
@fmlv.validate("<fn>", "FAIL"\|"WARN") |
รันการตรวจสอบความถูกต้องพร้อมผลลัพธ์ผ่าน/ไม่ผ่านและเลือกการดําเนินการ (ดู ควบคุมผลลัพธ์ ด้านล่าง) |
@fmlv.analyze("<fn>") |
รันการตรวจสอบโปรไฟล์ที่คํานวณเมตริกคอลัมน์ (ขนาด, ความสมบูรณ์, ค่าเฉลี่ย, ฯลฯ) ผลิตแค่ตัวชี้วัดเท่านั้น — ไม่เคยล้มเหลวในการรัน |
เนื่องจากชุดนี้มุ่งเป้าไปที่ MLV ตามชื่อ คุณภาพข้อมูลจึงสามารถเป็นเจ้าของ รัน และตั้งเวลาได้อย่างอิสระจากนิยาม MLV — และคุณสามารถรันการตรวจสอบโดยไม่ทําให้เกิดการรีเฟรช
ขั้นตอนที่ 1: ตั้งค่ารันไทม์
กําหนดค่าการพึ่งพาสําหรับเวอร์ชัน Fabric runtime ของคุณก่อนกําหนดการตรวจสอบคุณภาพข้อมูลระดับคอลัมน์
ระยะเวลาการใช้งานของ Fabric ก่อนเวอร์ชัน 2.0
ดาวน์โหลดวงล้อ PyDeequ 1.7.0 และอัปโหลดเป็นไลบรารีแบบกําหนดเองในสภาพแวดล้อม Fabric ของคุณ สําหรับคําแนะนํา ดูที่ อัปโหลดไลบรารีที่กําหนดเอง
Fabric runtime เวอร์ชัน 2.0 หรือใหม่กว่า
เพิ่มคุณสมบัติ Spark ต่อไปนี้:
| คุณสมบัติของ Spark | มูลค่า |
|---|---|
spark.driver.extraClassPath |
/opt/deequ/jars/* |
spark.executor.extraClassPath |
/opt/deequ/jars/* |
คุณสามารถเพิ่มคุณสมบัติเหล่านี้ได้ทั้งในสภาพแวดล้อมหรือสมุดบันทึก:
- สภาพแวดล้อม: เพิ่มคุณสมบัติภายใต้ Spark Compute>คุณสมบัติของ Spark จากนั้นบันทึกและเผยแพร่สภาพแวดล้อม สําหรับคําแนะนํา ดูที่ เพิ่มการกําหนดค่า Spark ไปยังสภาพแวดล้อมที่กําหนดเอง
-
สมุดบันทึก: เพิ่มคุณสมบัติใน
confวัตถุของ%%configureคําสั่งในเซลล์โค้ดแรก ก่อนเริ่มเซสชัน Spark สําหรับคําแนะนํา ดูคําสั่ง Magic การตั้งค่าเซสชัน Spark
ขั้นตอนที่ 2: กําหนดการตรวจสอบระดับคอลัมน์
ในตัวอย่างนี้ กําหนดการตรวจสอบคุณภาพข้อมูลระดับคอลัมน์สําหรับsales_db.daily_ordersมุมมองทะเลสาบที่สร้างขึ้นจริง โดยระบุโดย mlv_name ใน@fmlv.dq_suite
นําการลงทะเบียนชุดซอฟต์แวร์และฟังก์ชันการป้อนข้อมูลมารวมไว้ในเซลล์คุณภาพข้อมูลเดียว ตัวอย่างต่อไปนี้รัน การตรวจสอบสองครั้ง ในรอบการตรวจสอบเดียว — customer_id ต้องสมบูรณ์ (ไม่มีค่าว่าง) และ order_id ต้องไม่ซ้ํากัน:
# Data quality cell — separate from the MLV definition
@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
def daily_orders_dq():
pass
# Entry function — returns a PyDeequ VerificationResult DataFrame
def completeness_and_uniqueness(df):
from pydeequ.checks import Check, CheckLevel
from pydeequ.verification import VerificationSuite, VerificationResult
check = Check(spark, CheckLevel.Error, "Completeness & Uniqueness")
result = (
VerificationSuite(spark)
.onData(df)
.addCheck(
check.isComplete("customer_id") # check 1: no nulls
.isUnique("order_id") # check 2: no duplicates
)
.run()
)
return VerificationResult.checkResultsAsDataFrame(spark, result)
เพิ่มเมตริกโดยไม่ล้มเหลวในการรัน ซ้อน @fmlv.analyze ผ่านเพื่อคํานวณเมตริกคอลัมน์ (ขนาด, ความสมบูรณ์, ค่าเฉลี่ย ฯลฯ) เพื่อให้เห็นได้ชัดเจน ผลลัพธ์ของการวิเคราะห์จะถูกบันทึกไว้เสมอและไม่บล็อกการรัน — มีประโยชน์ในขณะที่กฎการตรวจสอบยังอยู่ระหว่างการพัฒนา
@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
@fmlv.analyze("column_profiling")
def daily_orders_dq():
pass
def column_profiling(df):
from pydeequ.analyzers import AnalysisRunner, AnalyzerContext, Size, Completeness, Mean
result = (
AnalysisRunner(spark)
.onData(df)
.addAnalyzer(Size())
.addAnalyzer(Completeness("customer_id"))
.addAnalyzer(Mean("amount"))
.run()
)
return AnalyzerContext.successMetricsAsDataFrame(spark, result)
ควบคุมผลลัพธ์: FAIL กับ WARN
การดําเนินการบน @fmlv.validate จะตัดสินว่าจะเกิดอะไรขึ้นเมื่อการตรวจสอบล้มเหลว:
-
FAIL— การตรวจสอบล้มเหลวทําให้การตรวจสอบคุณภาพข้อมูลล้มเหลวและบล็อก MLV ที่อยู่ปลายทางที่ขึ้นอยู่กับการตรวจสอบนั้น ใช้มันสําหรับสัญญาที่ยาก เช่น กุญแจหลักและฟิลด์บังคับ -
WARN— ความล้มเหลวถูกบันทึกไว้แต่การรันยังคงสําเร็จและ MLV ด้านล่าง ไม่ ถูกบล็อก ใช้สําหรับกฎแบบนุ่มนวลหรือกฎที่ปรึกษา
ขั้นตอนที่ 3: เปิดใช้งานการตรวจสอบคุณภาพข้อมูลในตารางเวลา
เลือกว่าการตรวจสอบระดับคอลัมน์จะทํางานเป็นส่วนหนึ่งของตารางรีเฟรชหรือไม่ ใน การตั้งค่าขั้นสูง ของประสบการณ์ การสร้างตารางเวลาใหม่ เมนูแบบเลื่อนลง การตรวจสอบคุณภาพข้อมูล มีตัวเลือกสองแบบ:
- ตรวจสอบหลังจากทุกครั้งที่รัน — ชุดคุณภาพข้อมูลจะทํางานโดยอัตโนมัติหลังจากการรีเฟรช MLV ในตารางเวลานั้นเสร็จสิ้น
- ไม่มีการตรวจสอบ — คุณภาพข้อมูลจะถูกข้ามสําหรับตารางเวลานั้น;MLV ยังคงรีเฟรชตามจังหวะของมัน
การตั้งค่าการตรวจสอบคุณภาพข้อมูลในการตั้งค่าขั้นสูงของตารางเวลา — เลือก ตรวจสอบหลังจากรันทุกครั้ง หรือ ไม่มีการตรวจสอบ
การตรวจสอบที่ล้มเหลวส่งผลต่อ pipeline อย่างไร
ในมุมมองสายสืบทอด การตรวจสอบคุณภาพข้อมูลถูก gold.daily_dq_reportusage กําหนดไว้และ FAIL ล้มเหลว ผลที่ตามมาคือสาขาที่ขึ้นกับ gold.monthly_dq_report…มัน , จะถูก ข้ามโดยอัตโนมัติ ในขณะที่สาขาอิสระ (gold.expected_regions, gold.region_mlv_dq_rep…) จะทํางานตามปกติ
FAILการตรวจสอบคุณภาพข้อมูลจะหยุดเชน — โหนดที่ล้มเหลวจะพึ่งพาการพึ่งพาด้านล่าง แต่สาขาอิสระยังคงดําเนินต่อไป
คําสั่งประหารชีวิต เมื่อชุดมีทั้งสองประเภท ฟังก์ชันทั้งหมด@fmlv.validateจะทํางานก่อน (ตามลําดับคํานิยาม) แล้วจึงทํางานฟังก์ชันทั้งหมด@fmlv.analyze การตรวจสอบที่ล้มเหลวจะถูกบันทึกและการดําเนินการจะดําเนินต่อไป;ผลลัพธ์โดยรวมจะถูกตัดสินหลังจากการตรวจสอบทั้งหมดเสร็จสิ้น — การรันจะล้มเหลวก็ต่อเมื่อ FAILการตรวจสอบ -action ล้มเหลวเท่านั้น
ขั้นตอนที่ 4: ดูและวิเคราะห์รายงานคุณภาพข้อมูล
ประวัติการรันจะแสดงว่าคุณภาพข้อมูลเปิดหรือปิดในแต่ละการรันงาน คุณยังสามารถรันชุดซอฟต์แวร์ตามความต้องการจากโน้ตบุ๊กได้ทุกเมื่อ โดยไม่ขึ้นกับการรีเฟรชใด ๆ
เลือกโหนด MLV ใดก็ได้ในมุมมองสายสืบทอด และเปิดแท็บ คุณภาพข้อมูล ในบานหน้าต่างรายละเอียดเพื่อดูผลลัพธ์ของรันนั้น:
-
Calculated Metrics — เมตริกรวมระดับคอลัมน์ที่คํานวณโดยเครื่องวิเคราะห์ Deequ (เช่น
Completenessของregion) - ผลการตรวจสอบ — สรุปจํานวน การตรวจสอบทั้งหมด จํานวนที่ ผ่านทั้งหมด และความ ล้มเหลวทั้งหมด ตามด้วยข้อจํากัดแต่ละข้อและ สถานะผ่าน/ไม่ผ่าน เพื่อให้คุณเห็นได้อย่างชัดเจนว่ากฎใดผ่านหรือไม่ผ่าน
แท็บ คุณภาพข้อมูล จะแสดงเมตริกที่คํานวณและผลการตรวจสอบต่อข้อจํากัดสําหรับรัน MLV ที่เลือก
ประวัติการรัน MLV ยังแสดงสถานะคุณภาพข้อมูลสําหรับแต่ละรัน — สําเร็จ, ล้มเหลว หรือไม่มีตัวบ่งชี้เมื่อไม่มีชุดซอฟต์แวร์รัน — เพื่อให้คุณสามารถสังเกตปัญหารันได้ทันทีโดยไม่ต้องเปิดโหนดแต่ละตัว
การตรวจสอบและเครื่องวิเคราะห์ที่รองรับ
เนื่องจาก DQ ระดับคอลัมน์ขับเคลื่อนโดย Amazon Deequ คุณสามารถใช้ไลบรารี Deequ ทั้งหมดภายในฟังก์ชันการป้อนข้อมูลของคุณได้
การตรวจสอบความถูกต้อง (ใช้ร่วมกับ @fmlv.validate) คือ Check ข้อจํากัดที่ให้ผลลัพธ์ผ่าน/ไม่ผ่าน; ตัววิเคราะห์ (ใช้ร่วมกับ @fmlv.analyze) คํานวณเมตริกโดยไม่มีผลผ่าน/ไม่ผ่าน ประเภทที่พบบ่อยที่สุดมีรายการในตารางต่อไปนี้
การตรวจสอบความถูกต้อง — @fmlv.validate
| เช็ค | คำอธิบาย | ตัวอย่าง |
|---|---|---|
hasSize |
จํานวนแถวของมุมมองเป็นไปตามเงื่อนไขหนึ่ง | .hasSize(lambda s: s == 5) |
isComplete |
คอลัมน์นี้ไม่มีค่า null | .isComplete("customer_id") |
hasCompleteness |
สัดส่วนของค่าที่ไม่เป็นศูนย์จะตรงตามเกณฑ์ที่กําหนด | .hasCompleteness("email", lambda c: c >= 0.9) |
isUnique |
คอลัมน์นี้ไม่มีค่าที่ซ้ํากัน | .isUnique("order_id") |
hasUniqueness |
สัดส่วนของค่าที่ไม่ซ้ํากันนั้นตรงตามเงื่อนไข | .hasUniqueness(["brand"], lambda u: u == 1.0) |
hasDistinctness |
สัดส่วนของค่าที่แตกต่างกันนั้นตรงตามเงื่อนไข | .hasDistinctness(["title"], lambda d: d > 0.5) |
isPrimaryKey |
คอลัมน์ (หรือชุดคอลัมน์) จะระบุแต่ละแถวอย่างเฉพาะเจาะจง | .isPrimaryKey("order_id") |
isContainedIn |
ค่าทั้งหมดอยู่ในชุดที่อนุญาต | .isContainedIn("priority", ["high", "low"]) |
isNonNegative / isPositive |
คอลัมน์นี้ไม่มีค่าลบ / มีแต่ค่าบวกอย่างเคร่งครัด | .isNonNegative("num_views") |
isLessThan / isGreaterThan |
คอลัมน์หนึ่งมักจะน้อยกว่าหรือมากกว่าคอลัมน์อื่นอย่างสม่ําเสมอ | .isLessThan("discount", "price") |
hasMin / hasMax |
คอลัมน์ขั้นต่ํา / สูงสุด เป็นไปตามเงื่อนไข | .hasMax("age", lambda v: v <= 120) |
hasMean / hasSum |
ค่าเฉลี่ย / ผลรวมของคอลัมน์ตรงตามเงื่อนไข | .hasSum("total_votes", lambda v: v > 0) |
hasStandardDeviation |
ส่วนเบี่ยงเบนมาตรฐานของคอลัมน์เป็นไปตามเงื่อนไข | .hasStandardDeviation("amount", lambda v: v < 5) |
hasApproxQuantile |
ควอนไทล์โดยประมาณ (เช่น ค่ามัธยฐาน) จะตรงตามเงื่อนไข | .hasApproxQuantile("num_views", 0.5, lambda v: v <= 10) |
hasCorrelation |
ความสัมพันธ์เพียร์สันระหว่างสองคอลัมน์มีเงื่อนไข | .hasCorrelation("a", "b", lambda c: c > 0.8) |
hasNumberOfDistinctValues |
จํานวนค่าที่แตกต่างกันตรงตามเงื่อนไข | .hasNumberOfDistinctValues("locale", lambda n: n == 5) |
hasEntropy |
เนื้อหาข้อมูล (เอนโทรปี) ของคอลัมน์ตรงตามเงื่อนไข | .hasEntropy("locale", lambda e: e > 2.0) |
hasDataType |
ค่าต่าง ๆ เป็นไปตามชนิดข้อมูลที่คาดหวัง | .hasDataType("id", ConstrainableDataTypes.Integral) |
hasPattern / containsURL / containsEmail |
สัดส่วนของค่าที่ตรงกับรูปแบบ regex / URL / อีเมลจะตรงตามเงื่อนไข | .containsURL("description", lambda c: c >= 0.5) |
containsCreditCardNumber / containsSocialSecurityNumber |
สัดส่วนของมูลค่าที่ตรงกับรูปแบบบัตรเครดิต / หมายเลขประกันสังคมจะตรงตามเงื่อนไข | .containsCreditCardNumber("notes", lambda c: c == 0) |
satisfies |
สัดส่วนของแถวที่เป็นไปตามพรีดิเคต SQL ที่กําหนดเองจะตรงตามเงื่อนไข | .satisfies("age >= 18", "adults", lambda c: c == 1.0) |
ตัววิเคราะห์ / ตัวชี้วัด — @fmlv.analyze
| เมตริก | คำอธิบาย | ตัวอย่าง |
|---|---|---|
ApproxCountDistinct |
จํานวนค่าที่แตกต่างกันโดยประมาณ คํานวณด้วยสเก็ตช์ HyperLogLog++ | ApproxCountDistinct("product_id") |
ApproxQuantile |
ควอนไทล์โดยประมาณของการแจกแจง | ApproxQuantile("locale_encoded", 0.5) |
ApproxQuantiles |
ควอนไทล์โดยประมาณหลายตัวของการแจกแจง | ApproxQuantiles("locale_encoded", [0.1, 0.5, 0.9]) |
Completeness |
เศษส่วนของค่าที่ไม่เป็นศูนย์ในคอลัมน์ | Completeness("product_id") |
Compliance |
สัดส่วนของแถวที่เป็นไปตามข้อจํากัดของคอลัมน์ที่กําหนด | Compliance("locale", "locale in ('es','jp','us')") |
Correlation |
ความสัมพันธ์เพียร์สันระหว่างสองคอลัมน์ ในช่วง [-1, 1] | Correlation("title", "brand") |
CountDistinct |
จํานวนค่าที่แตกต่างกัน | CountDistinct("locale") |
DataType |
การแจกแจงของชนิดข้อมูล (บูลีน, เศษส่วน, อินทิกรัล, สตริง) | DataType("locale") |
Distinctness |
เศษส่วนของค่าที่แตกต่างกันบนจํานวนค่าทั้งหมด | Distinctness("title") |
Entropy |
เนื้อหาข้อมูลของการแจกแจงค่าของคอลัมน์ วัดเป็น nats | Entropy("locale_encoded") |
Maximum / Minimum |
ค่าสูงสุด / ต่ําสุดของคอลัมน์ | Maximum("locale_encoded") |
Mean |
ค่าเฉลี่ย;ค่าที่เป็นศูนย์จะถูกตัดออก | Mean("locale_encoded") |
MutualInformation |
สามารถอนุมานข้อมูลเกี่ยวกับคอลัมน์หนึ่งจากอีกคอลัมน์หนึ่งได้มากน้อยเพียงใดสมมาตรและไม่เป็นลบ | MutualInformation(["title", "brand"]) |
PatternMatch |
สัดส่วนของแถวที่สอดคล้องกับนิพจน์ปกติที่กําหนด | PatternMatch("locale", r"\w{2}") |
Size |
จํานวนแถวใน DataFrame | Size() |
Sum |
ผลรวมของค่าทั้งหมดในคอลัมน์ | Sum("total_votes") |
UniqueValueRatio |
เศษส่วนของค่าที่ไม่ซ้ํากัน แบ่งตามจํานวนค่าที่แตกต่างกันทั้งหมด | UniqueValueRatio(["brand"]) |
Uniqueness |
สัดส่วนของค่าที่ไม่ซ้ํากัน แบ่งเป็นจํานวนค่าทั้งหมด | Uniqueness(["brand"]) |
ไลบรารี Amazon Deequ แบบโอเพ่นซอร์สมีการตรวจสอบและวิเคราะห์เหล่านี้ สําหรับรายชื่อที่ครบถ้วนและเป็นที่เชื่อถือ โปรดดูโครงการ Deequ ที่ github.com/awslabs/deequ
Summary
- สร้าง MLV ของคุณและ จดชื่อเต็มของมันไว้
-
ในเซลล์คุณภาพข้อมูลแยกต่างหาก ให้ ลงทะเบียนชุดข้อมูลที่มี
@fmlv.dq_suite(mlv_name=…). -
เพิ่มการตรวจสอบ — stack
@fmlv.validate("<fn>", "FAIL"|"WARN")และ/or@fmlv.analyze("<fn>")และเขียนฟังก์ชันรายการที่ตรงกันโดยคืนค่า PyDeequ DataFrames - เรียกใช้ — ตามคําสั่งจากสมุดบันทึก หรือเปิดสวิตช์คุณภาพข้อมูลเมื่อสร้างตารางเวลา
- ตรวจสอบผลลัพธ์ ในแท็บ lineage คุณภาพข้อมูล และประวัติการรัน
เนื้อหาที่เกี่ยวข้อง
- คุณภาพข้อมูลใน Materialized Lake Views — Microsoft Learn
- ภาพรวมวิวทะเลสาบที่สร้างขึ้น — Microsoft Learn
- เอกสาร DeeQu