Tutorial Great Expectations: Validasi Data dengan Python

Menyiapkan Great Expectations

Dalam tutorial ini, Anda akan belajar cara menggunakan GX Core, versi open-source dari Great Expectations, untuk memvalidasi sebuah Pandas DataFrame. Kita akan membahas penyiapan context, mendaftarkan sumber data Pandas, mendefinisikan expectations, dan memvalidasi batch data.

Catatan: Kami menyarankan Anda mengikuti notebook DataLab, tetapi Anda juga bisa membuat skrip Python sendiri.

1. Menginstal Great Expectations

Prasyarat

• Python 3.9 hingga 3.12 sudah terpasang.
• Untuk menghindari konflik, sangat disarankan memasang Great Expectations di dalam lingkungan virtual (disclaimer: penyiapan lingkungan virtual tidak dibahas dalam artikel ini).
• Dataset contoh.

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

Gunakan perintah berikut untuk memasang GX melalui pip:

pip install great_expectations

Perintah ini memasang paket inti beserta seluruh dependensi yang diperlukan.

2. Menginisialisasi data context

Great Expectations memerlukan data context untuk mengelola konfigurasi. Kami menggunakan ephemeral data context agar tidak menyimpan konfigurasi secara permanen.

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

Sekarang GX sudah siap, mari 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 file).

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 mengorganisasi 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 kumpulan 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. Pada contoh ini, kita akan mendefinisikan expectations sederhana berikut:

1. Pastikan nilai inventory_id tidak 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 menjelajahi semua expectations yang tersedia di Expectation Gallery. Kami mendorong Anda untuk 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 dan ID suite: Nama unik (inventory_parts_suite) dan pengenal untuk melacak serta mengelola suite.
2. Expectations: Setiap aturan menentukan:
• Jenis pemeriksaan (misalnya memastikan sebuah kolom tidak memiliki nilai null atau entri harus unik).
• Parameter, seperti kolom yang divalidasi.
• Metadata dan ID unik untuk setiap expectation, memudahkan pelacakan dan kustomisasi.
3. Metadata: Informasi versi Great Expectations untuk memastikan kompatibilitas 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 jelas didefinisikan, dapat ditelusuri, dan siap digunakan kembali.

5. Memvalidasi data

Terakhir, validasi batch terhadap expectations yang telah ditentukan 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 mengenai 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 sederhana dari 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 menunjukkan seberapa sering duplikasi tersebut muncul dan posisi barisnya, memudahkan Anda 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 dalam alur kerja untuk memantau kualitas data secara berkelanjutan di setiap tahap.

Di bagian ini, kita 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 setiap alat untuk sintaks yang terbaru!

Integrasi dengan alat ETL

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

Mari kita telusuri contoh sederhana mengintegrasikan Great Expectations dengan Prefect untuk menjalankan validasi data sebagai bagian dari workflow 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()

Pada 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 menimbulkan peringatan, menghentikan workflow dan mencegah kesalahan di tahap berikutnya.

Validasi data berkelanjutan

Anda dapat menjadwalkan pekerjaan validasi menggunakan berbagai alat, seperti cron job di sistem berbasis Unix atau layanan terkelola seperti Apache Airflow. Untuk contoh ini, kita akan mendemonstrasikan cara menjadwalkan menjalankan 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 menjalankan validasi sekali sehari (@daily).

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

Praktik Terbaik untuk Validasi Data dengan Great Expectations

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

Mulai dari yang kecil dan lakukan iterasi

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 tentang dataset meningkat, Anda dapat menambahkan validasi yang lebih kompleks.

Berkolaborasi lintas tim

Kualitas data bukan hanya masalah teknis. Berkolaborasilah lintas 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 memenuhi kebutuhan semua pemangku kepentingan.

Otomatiskan jika memungkinkan

Otomatisasikan proses sejauh mungkin untuk mengintegrasikan validasi data ke dalam pipeline data. Mengintegrasikan 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 data yang tinggi dan transparansi dalam alur kerja Anda.

Untuk terus mengembangkan keterampilan Anda, lihat sumber berikut:

• ETL and ELT in Python: Pelajari cara mentransformasikan 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 ringkas dimensi kualitas data.