Data Pipeline larda malumotlarni tekshirish (validatsiya qilish) juda muhim bolib, u pipeline jarayonining turli bosqichlarida amalga oshiriladi.

data analyst lar uchun pipeline larda validatsiya odatda ma'lumotlarni olish (ingestion, extract-load) bosqichidan keyin va malumotlarni modellashtirish (transform) jarayonidan song amalga oshiriladi.
ML pipeline larida esa, yuklab olingan malumotlarni validatsiya qilish ayniqsa muhim. Bu bosqichda notogri yoki sifatsiz malumotlarni aniqlash va yoq qilish amalga oshiriladi.
Muhimi, malumot validatsiyasini ML modelning validatsiyasi bilan aralashtirmaslik kerak. Malumot validatsiyasi kiruvchi malumotlarning sifatini tekshirishga qaratilgan bolsa, ML model validatsiyasi esa modelning ishlash samaradorligini baholash jarayonidir.

import configparser
import pymysql
import csv
import boto3

# Config faylini oqish
parser = configparser.ConfigParser()
parser.read("pipeline.conf")

# MySQL ulanish malumotlarini olish
hostname = parser.get("mysql_config", "hostname")
port = parser.get("mysql_config", "port")
username = parser.get("mysql_config", "username")
dbname = parser.get("mysql_config", "database")
password = parser.get("mysql_config", "password")

# MySQL bazasiga ulanish
conn = pymysql.connect(
    host=hostname,
    user=username,
    password=password,
    db=dbname,
    port=int(port)
)

if conn is None:
    print("Error connecting to the MySQL database")
else:
    print("MySQL connection established!")

# Orders jadvalidan malumotlarni olish va CSV faylga yozish
m_query = "SELECT * FROM Orders;"
local_filename = "order_extract.csv"
m_cursor = conn.cursor()
m_cursor.execute(m_query)
results = m_cursor.fetchall()

with open(local_filename, 'w') as fp:
    csv_w = csv.writer(fp, delimiter='|')
    csv_w.writerows(results)

fp.close()
m_cursor.close()
conn.close()

# S3 ga yuklash uchun AWS credentials malumotlarini olish
parser = configparser.ConfigParser()
parser.read("pipeline.conf")

access_key = parser.get("aws_boto_credentials", "access_key")
secret_key = parser.get("aws_boto_credentials", "secret_key")
bucket_name = parser.get("aws_boto_credentials", "bucket_name")

# S3 bilan ulanish
s3 = boto3.client(
    's3',
    aws_access_key_id=access_key,
    aws_secret_access_key=secret_key
)

# CSV faylni S3 ga yuklash
s3_file = local_filename
s3.upload_file(local_filename, bucket_name, s3_file)

import configparser
import psycopg2

# Config faylini oqish
parser = configparser.ConfigParser()
parser.read("pipeline.conf")

# Redshift ulanish malumotlarini olish
dbname = parser.get("aws_creds", "database")
user = parser.get("aws_creds", "username")
password = parser.get("aws_creds", "password")
host = parser.get("aws_creds", "host")
port = parser.get("aws_creds", "port")

# Redshiftga ulanish
rs_conn = psycopg2.connect(
    "dbname=" + dbname
    + " user=" + user
    + " password=" + password
    + " host=" + host
    + " port=" + port
)

# Orders jadvalidagi songgi yangilangan sanani olish
rs_sql = """SELECT COALESCE(MAX(LastUpdated), '1900-01-01') FROM Orders;"""

rs_cursor = rs_conn.cursor()
rs_cursor.execute(rs_sql)
result = rs_cursor.fetchone()

# Natijani olish
last_updated_warehouse = result[0]

# Kursorni yopish
rs_cursor.close()

SELECT * FROM Orders WHERE LastUpdated > '2024-02-10';

from pendulum import DateTime

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def sender1():
        # 'data' key ostida brand qiymatini yuborish
        ti.xcom_push(key='data', value={'brand': 'APPLE'})

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def sender2():
        # 'return_value' keyi avtomatik tarzda ishlatiladi
        return {'brand': 'Apple'}

from airflow import DAG
from airflow.decorators import task

with DAG(...) as dag:

    @task()
    def receiver1():
        # sender task'dan 'data' keyidagi qiymatni olish
        ti.xcom_pull(task_ids='sender', key='data')

sender() >> receiver()  

receiver(sender())