from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
train = KubernetesPodOperator(
    task_id='train_model',
    name='train-model-{{ ds_nodash }}',
    namespace='airflow',
    image='myrepo/ml-train:1.0.0',
    cmds=['python', 'train.py'],
    arguments=['--epochs', '20'],
    resources={'request_memory': '2Gi', 'request_cpu': '1000m'},
    get_logs=True,
    is_delete_operator_pod=True
)

TRUNCATE TABLE customers;
INSERT INTO customers
SELECT * FROM staging.customers;

-- Oxirgi yuklangan sana saqlanadi
SELECT * FROM source_table
WHERE updated_at > '2025-04-27 23:59:59';

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("streaming_app").getOrCreate()
df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "topic_name").load()

from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.microsoft.mssql.operators.mssql import MsSqlOperator
from datetime import datetime

with DAG(dag_id="full_load_customers",
         start_date=datetime(2024, 1, 1),
         schedule_interval="@daily",
         catchup=False) as dag:

    truncate_target_table = MsSqlOperator(
        task_id='truncate_target_table',
        mssql_conn_id='mssql_target',
        sql="TRUNCATE TABLE customers;"
    )

    load_full_data = PostgresOperator(
        task_id='load_full_data',
        postgres_conn_id='postgres_source',
        sql="""
            INSERT INTO mssql_target.dbo.customers
            SELECT * FROM postgres_source.public.customers;
        """
    )

    truncate_target_table >> load_full_data