Tutorial Great Expectations: Memvalidasi Data dengan Python

Menyiapkan Great Expectations

Dalam tutorial ini, Anda akan mempelajari cara menggunakan GX Core, versi open-source dari Great Expectations, untuk memvalidasi Pandas DataFrame. Kita akan membahas penyiapan context, registrasi sumber data Pandas, pendefinisian expectations, dan validasi batch data.

Catatan: Kami merekomendasikan Anda mengikuti notebook DataLab, tetapi Anda juga dapat membuat skrip Python sendiri.

1. Menginstal Great Expectations

Prasyarat

• Python 3.9 hingga 3.12 terpasang.
• Untuk menghindari konflik, sangat disarankan agar Anda menginstal Great Expectations di dalam lingkungan virtual (disclaimer: penyiapan lingkungan virtual tidak dibahas dalam artikel ini).
• Satu set data contoh.

Catatan: Jika menggunakan notebook DataLab yang disediakan, prasyarat ini sudah terpenuhi. Silakan lewati.

Gunakan perintah berikut untuk menginstal GX melalui pip:

pip install great_expectations

Perintah ini menginstal paket inti dan semua dependensi yang diperlukan.

2. Menginisialisasi data context

Great Expectations memerlukan data context untuk mengelola konfigurasi. Kita menggunakan ephemeral data context agar konfigurasi tidak dipersistenkan.

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Membuat Suite Validasi Data Pertama Anda

Setelah GX siap, mari kita buat suite validasi data.

1. Menghubungkan ke sumber data dan membuat data asset

Sumber data menghubungkan Great Expectations ke data Anda, sementara data asset merepresentasikan subset data tertentu (misalnya tabel, DataFrame, atau berkas).

Dalam kasus ini, kita akan menyiapkan semuanya untuk terhubung ke DataFrame bernama inventory_parts_df. Dataset contoh tersedia di DataLab yang disediakan, dan akan dibuat setelah kita menjalankan blok SQL:

Jika Anda tidak menggunakan DataLab, buatlah DataFrame Anda sendiri dengan data contoh.

Sekarang, buat sumber data dan asset Anda:

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Menambahkan definisi batch

Definisi batch mengidentifikasi dan mengatur data Anda untuk validasi. Di sini, kita menambahkan definisi batch yang mencakup seluruh DataFrame:

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Mengambil batch

Batch adalah sekumpulan data yang terikat pada definisi batch. Untuk memvalidasi data, Anda perlu mengambil dan menautkan batch ke DataFrame Anda, dalam hal ini inventory_parts_df:

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Membuat suite dan mendefinisikan expectations

Expectations adalah aturan untuk memvalidasi data. Dalam contoh ini, kita akan mendefinisikan expectations sederhana berikut:

1. Pastikan nilai inventory_id tidak bernilai null.
2. Pastikan nilai part_num unik.

# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

Anda dapat mengeksplor semua expectations yang tersedia di Expectation Gallery. Kami mendorong Anda menambahkan beberapa lagi!

Setelah mendefinisikan expectations, GX menghasilkan konfigurasi expectation suite:

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

Suite ini mencakup detail berikut:

1. Nama suite dan ID: Nama unik (inventory_parts_suite) dan pengenal untuk melacak dan mengelola suite.
2. Expectations: Setiap aturan menetapkan:
• Jenis pemeriksaan (misalnya memastikan kolom tidak memiliki nilai null atau memiliki entri unik).
• Parameter, seperti kolom yang divalidasi.
• Metadata dan ID unik untuk setiap expectation, memudahkan pelacakan dan kustomisasi.
3. Metadata: Informasi versi Great Expectations, memastikan kompatibilitas dengan alat.
4. Catatan: Placeholder untuk menambahkan komentar deskriptif tentang suite (opsional).

Output terstruktur ini berfungsi sebagai dokumentasi sekaligus konfigurasi yang dapat digunakan kembali untuk memvalidasi dataset Anda sehingga expectations Anda terdefinisi jelas, dapat dilacak, dan siap digunakan di masa mendatang.

5. Memvalidasi data

Terakhir, validasi batch terhadap expectations yang telah ditetapkan dan evaluasi hasilnya.

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

Setelah menjalankan validasi, Great Expectations memberikan laporan terperinci apakah dataset memenuhi expectations yang ditetapkan:

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

Laporan ini merinci kualitas data Anda, menyoroti keberhasilan dan kegagalan. Berikut penjelasan ringkas hasilnya:

Validasi keseluruhan: Hasil validasi sebagian berhasil: 50% expectation lolos, dan 50% gagal. Expectation yang gagal menunjukkan masalah kualitas data yang perlu diperhatikan. Dalam kasus ini, satu kolom tidak memenuhi aturan yang ditetapkan.

Expectation 1: inventory_id tidak boleh memiliki nilai hilang

• Hasil: Lolos
• Penjelasan: Setiap nilai pada kolom inventory_id ada, tanpa entri null atau hilang. Ini menunjukkan kelengkapan data yang baik untuk kolom ini.

Expectation 2: part_num harus memiliki nilai unik

• Hasil: Gagal
• Penjelasan: Kolom part_num berisi 97,98% nilai duplikat, artinya hanya sedikit nilai yang unik.
• Sorotan:
• Contoh nilai duplikat termasuk "3069b" dan "33291".
• Alat ini juga menampilkan seberapa sering duplikat tersebut muncul dan posisi barisnya, sehingga lebih mudah menemukan dan memperbaiki masalah.

Tentu saja, ini hanya dataset contoh, dan kami sengaja menyertakan satu expectation yang lolos dan satu yang gagal agar Anda dapat melihat kedua hasil validasi.

Selesai! Anda telah berhasil menjalankan validasi data end-to-end.

Mengintegrasikan Great Expectations ke dalam Pipeline Data

Dalam lingkungan produksi, validasi harus ditanamkan langsung ke alur kerja untuk memantau kualitas data secara berkelanjutan di setiap tahap.

Pada bagian ini, kami akan membahas cara mengintegrasikan Great Expectations ke dalam pipeline data Anda.

Ini adalah contoh untuk memberi gambaran, dan mungkin diperlukan konfigurasi tambahan yang tidak disertakan di sini. Lihat dokumentasi masing-masing alat untuk sintaks terbaru!

Integrasi dengan alat ETL

Mengintegrasikan Great Expectations dengan alat ETL populer seperti Apache Airflow atau Prefect relatif mudah. Menanamkan langkah validasi langsung ke proses ETL memungkinkan Anda mendeteksi dan menangani masalah data secara real time sebelum memengaruhi analisis hilir.

Mari kita bahas contoh sederhana mengintegrasikan Great Expectations dengan Prefect untuk menjalankan validasi data sebagai bagian dari alur kerja ETL otomatis:

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

Dalam contoh ini, kita mendefinisikan flow Prefect dengan sebuah task untuk menjalankan validasi Great Expectations.

Fungsi validate_data() memuat context Great Expectations, mengambil batch data, dan menerapkan expectation suite.

Jika data tidak memenuhi kriteria validasi, task akan memicu peringatan, menghentikan alur kerja, dan mencegah kesalahan di tahap berikutnya.

Validasi data berkelanjutan

Anda dapat menjadwalkan job validasi menggunakan berbagai alat, seperti cron job pada sistem berbasis Unix atau layanan terkelola seperti Apache Airflow. Untuk contoh ini, kami akan menunjukkan cara menjadwalkan validasi menggunakan Airflow, yang cocok untuk mengorkestrasi pipeline data.

Berikut cara menyiapkan DAG Airflow (Directed Acyclic Graph) untuk menjalankan validasi Great Expectations setiap hari:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

Dalam contoh ini, kita mendefinisikan DAG yang menjadwalkan validasi sekali sehari (@daily).

Fungsi run_validation() mengeksekusi validasi dengan memuat context Great Expectations dan menjalankan expectation suite yang ditetapkan terhadap data.

Praktik Terbaik untuk Validasi Data dengan Great Expectations

Mengikuti praktik terbaik selalu dianjurkan untuk skalabilitas dan efisiensi, dan hal ini juga berlaku untuk validasi data dengan Great Expectations.

Mulai dari yang kecil dan iterasikan

Mulailah dengan pemeriksaan kualitas data dasar dan perluas secara bertahap. Lebih baik fokus pada expectations dasar terlebih dahulu untuk menghindari proses yang terlalu rumit, sehingga integrasi lebih mulus dan pemecahan masalah lebih mudah. Seiring pemahaman Anda terhadap dataset meningkat, Anda dapat menambahkan validasi yang lebih kompleks.

Ber kolaborasi lintas tim

Kualitas data bukan hanya masalah teknis. Berkolaborasilah dengan tim bisnis untuk mendefinisikan expectations dan memastikan validasi yang diterapkan selaras dengan logika dan tujuan bisnis yang mendasarinya. Pendekatan lintas fungsi ini menjamin data memenuhi tujuan yang dimaksud dan kebutuhan seluruh pemangku kepentingan.

Otomatisasi sebisa mungkin

Otomatiskan proses sebisa mungkin untuk mengintegrasikan validasi data ke dalam pipeline data. Integrasi pemeriksaan validasi otomatis memungkinkan pemantauan kualitas data secara berkelanjutan tanpa intervensi manual, yang secara signifikan meningkatkan efisiensi.

Kesimpulan

Kerja bagus! Anda telah mempelajari cara mengonfigurasi dan memvalidasi data di Great Expectations. Teknik-teknik ini akan membantu menjaga kualitas dan transparansi data yang tinggi dalam alur kerja Anda.

Untuk terus membangun keterampilan Anda, lihat sumber daya berikut:

• ETL and ELT in Python: Pelajari cara mentransformasi dan memindahkan data secara efektif.
• Introduction to Data Quality: Jelajahi dasar-dasar manajemen kualitas data.
• Cleaning Data in Python: Kuasai teknik pembersihan data untuk memastikan akurasi dan konsistensi.
• Data Quality Dimensions Cheat Sheet: Panduan praktis untuk dimensi kualitas data.