Airflow ile Bir ETL Boru Hattı Oluşturma

Apache Airflow kullanarak ETL boru hatları dünyasına hoş geldiniz. Bu derste, Polygon API’sini kullanarak borsa verilerini çekmeye, bu verileri dönüştürmeye ve ardından kolay erişim ve işleme için bir SQLite veritabanına yüklemeye odaklanacağız. Haydi başlayalım!

Apache Airflow ve Airflow ETL nedir?

Apache Airflow, veri orkestrasyonu ve boru hattı yönetimi için sektörde standart kabul edilir. Karmaşık iş akışlarını orkestre etme, görevler arasındaki bağımlılıkları yönetme, başarısız görevleri yeniden deneme ve kapsamlı günlükleme sağlama becerileri sayesinde veri bilimciler, makine öğrenimi mühendisleri ve yapay zeka uzmanları arasında popüler hale gelmiştir.

Airflow ETL, ETL süreçlerini yönetmek için Apache Airflow’un kullanılmasını ifade eder. Kısaca hatırlatmak gerekirse ETL; verilerin çeşitli kaynaklardan çıkarılması, analiz için uygun bir formata dönüştürülmesi ve bir veri ambarı gibi nihai bir hedefe yüklenmesini içeren bir veri entegrasyonu türüdür. 

Airflow Geliştirme Ortamımızı Yapılandırma 

Airflow ile bir ETL boru hattı oluşturmaya başlamadan önce geliştirme ortamımızı yapılandırmamız gerekir. Geliştirme ortamımızı nasıl yapılandıracağımıza dair ayrıntılı bilgi için Apache Airflow’a Başlarken öğreticimize göz atın.

Ayrıca Astro CLI’yi de kurmamız gerekecek. Astro CLI’yi geliştiren ve ayrıntılı belgeler sunan Astronomer’a kurulum talimatları için göz atın.

Bir Airflow Projesi Oluşturma

Ortamımızı yapılandırıp Astro CLI’yi kurduktan sonra bir Airflow projesi oluşturuyoruz. Bunu, bir terminal kabuğu açıp seçtiğimiz yolda yeni bir dizin oluşturarak yapıyoruz.

 ~/Documents/data-engineering/ETL-pipeline/ Altyapı:

Gerekli kaynakları oluşturmak için bu dizinin kökünden aşağıdaki komutu çalıştırıyoruz:

astro dev initAltyapı:

Dizinin içeriği aşağıdakine benzer görünecektir. Tam çıktı farklılık gösterebilir.

├── dags/
├── include/
├── plugins/
├── tests/
├── airflow_settings.yaml
├── Dockerfile
├── packages.txt
└── requirements.txtAltyapı:

Projenizi çalışır duruma getirmek için şu komutu yürütün: 

astro dev startAltyapı:

Airflow ortamınızın ayağa kalkması yaklaşık bir dakika sürecektir. Ardından web tarayıcınızda localhost: 8080 adresine gidin; Airflow arayüzü sizi karşılayacaktır.

Artık Airflow ile kendi ETL boru hattınızı geliştirmeye hazırsınız!

Bir ETL boru hattı tasarlama

Tek satır kod yazmadan önce boru hattınızın her bileşenini planlamak için zaman ayırmak önemlidir. Özellikle, önce veri kaynağınızı ve verilerinizin yükleneceği hedefi belirlemek iyi bir alışkanlıktır. Kaynakları ve hedefi önceden belirleyerek, verinin yol boyunca nasıl dönüştürüleceğine dair de fikir ediniriz.

Örneğimizde, Polygon API’sinden borsa verilerini çekip dönüştürdükten sonra bu verileri bir SQLite veritabanına yükleyecek bir veri boru hattı tasarlayacağız. Bu durumda, kaynak sistem Polygon API, hedef ise bir SQLite veritabanıdır. Bunu bir görselle anlatalım: 

Kaynak ve hedef diyagramı

Veri mühendisleri olarak deneyimimizden biliyoruz ki verilerin bir SQLite veritabanına yüklenmeye hazır hâle gelmesi için JSON’dan tabular bir formata dönüştürülmesi gerekir. Polygon API’den çekildikten sonra verilerimizi yerel Python ve pandas kütüphanesini kullanarak dönüştürmek üzere bir plan oluşturuyoruz. Görselimizi bu değişikliği gösterecek şekilde güncelleyelim:

Mimari diyagram

Bu ek bilgiyi ekleyerek sistemimizin geniş bir görsel temsili olan bir mimari diyagram oluşturduk. Bu boru hattındaki üç mantıksal adımı, yani sürecimizin E, T ve L’sini görüyoruz.

Ayrıca bunların her birini Airflow’un çalıştıracağı tüm görevleri, sıralarını ve birbirlerine bağımlılıklarını tanımlayan özel bir yapılandırma olan yönlendirilmiş çevrimsiz bir grafik (DAG) hâline de getirebiliriz. Yönlendirilmiş çevrimsiz grafikler hakkında daha fazla bilgi için Airflow DAG’larını ayrıntılı olarak inceleyen Veri Mühendisliğine Giriş Kursumuzu alabilirsiniz.

Kurumsal veri ortamlarında birçok ekip, tasarım kararları üzerinde uzlaşmak ve bunları belgelemek için teknik şartname (tech spec) adı verilen başka tür belgeler kullanır. Bizim için, veri boru hattımızın tasarım seçimlerini belgelemek üzere bir tablo oluşturmak yeterlidir. 

Veri boru hattı tercihleri

Unutmayın, DAG’ımız görev düzeyindeki ayrıntılara odaklanır ve her bilgiyi kapsamıyor olabilir. Çözmemiz gereken ek ayrıntıları belgelemek için ikinci bir tablo kullanabiliriz. Sorularımız şunları içerir:

• Bu DAG ne sıklıkla çalıştırılacak? 
• Boru hattındaki bir görev başarısız olursa ne olur? 
• Ek hisse senetleri hakkında veri toplamak istersek ne olur?

Netleştirilecek sorular

Başardıklarımızı gözden geçirelim: Bir mimari diyagram oluşturduk, boru hattının Airflow görevlerine nasıl ayrılacağını belgelendirdik ve DAG’ı yapılandırmak için gereken üst düzey bilgileri belirledik.

Veri boru hatları tasarlama, geliştirme ve test etme hakkında daha fazla bilgi edinmek isterseniz DataCamp’in Veri Boru Hatlarına Giriş kursuna göz atın. Burada Python ile ETL boru hatları kurmanın temellerini ve çözümünüzün sağlam, dayanıklı ve yeniden kullanılabilir olmasını sağlayan en iyi uygulamaları öğreneceksiniz.

Airflow ile Bir ETL Boru Hattı Oluşturma

ETL boru hattımızı adımları sırayla takip ederek inşa edeceğiz. Yapısal bir yaklaşım, her aşamanın hassasiyetle yürütülmesini sağlar. 

Airflow ile veri çıkarma

Polygon API’den veri çekmeden önce, Polygon’u ziyaret edip API Key Oluştur düğmesini seçerek bir API belirteci oluşturmamız gerekecek. Bu öğretici için ücretli bir aboneliğe kaydolmanıza gerek olmadığını lütfen unutmayın; ücretsiz katman bir API anahtarı üretir ve ihtiyacımız olan tüm işlevleri sağlar. API anahtarınızı kopyalayıp kaydetmeyi unutmayın.

API anahtarını oluşturduktan sonra, artık Airflow ile Polygon API’sinden veri çıkarmaya başlayabiliriz. DAG yapılandırmamıza ilişkin ayrıntıları içeren teknik şartname tablomuzu kod yazarken referans alacağız. 

from airflow import DAG 
from datetime import datetime, timedelta

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:Altyapı:

İlk görevimizi ekleyelim. TaskFlow API’yi ve requests modülünü kullanarak Polygon API’sinden hisse senedi açılış-kapanış verilerini çekeceğiz. 

import requests
...
@task()

def hit_polygon_api(**context):
    # Instantiate a list of tickers that will be pulled and looped over
    stock_ticker = "AMZN"
    # Set variables
    polygon_api_key= "<your-api-key>"
  ds = context.get("ds")

  # Create the URL
    url= f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
    response = requests.get(url)
    # Return the raw data
    return response.json()Altyapı:

Bu kodla ilgili dikkat edeceğiniz birkaç nokta var. 

• hit_polygon_api fonksiyonu @task ile dekore edilmiştir. Bu dekoratör, fonksiyonu bir DAG’ın parçası olarak yürütülebilecek bir Airflow görevi hâline getirir.
• hit_polygon_api imzasında context parametresi tanımlıdır. Daha sonra ds anahtarında saklanan değeri çıkarmak için kullanılır.
• context, görev ve DAG hakkında meta veri içeren bir sözlüktür.
• context sözlüğünden ds değerini çekerken, data_interaval_end tarihini YYYY-mm-dd biçiminde elde ederiz.  
• DAG çalıştırıldığında yeni görevimizin yürütüldüğünden emin olmak için hit_polygon_api çağrısını eklememiz gerekir.

Tüm bunları bir araya getirdiğimizde, ETL boru hattımızın ilk bölümünü oluşturacak kod şöyle görünür.

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta import requests

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=False,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    hit_polygon_api()Altyapı:

Bunu 1 Ocak 2024 için çalıştırdığınızda yanıtın bir hata içereceğini fark edeceksiniz. Piyasa o gün kapalı olduğundan, Polygon bir istisna belirten bir yanıt döndürür. Bunu bir sonraki adımda ele alacağız.

Airflow ile verileri dönüştürme

Polygon API’den veri çıkarıldıktan sonra, dönüştürmeye hazırız.

Bunu yapmak için TaskFlow API’yi kullanarak başka bir görev oluşturacağız. Bu görevin adı flatten_market_data olacak ve parametreleri polygon_response ( hit_polygon_api fonksiyonunun döndürdüğü ham veri) ile **context olacak. polygon_response’a birazdan yakından bakacağız.

Yapacağımız gerçek dönüşüm oldukça basit. Polygon API’nin döndürdüğü JSON’u bir listeye düzleştireceğiz. Püf noktası, yanıtta bulunmayan her anahtar için benzersiz varsayılan değerler sağlamamızdır.

Örneğin, yanıtta from anahtarı yoksa, Airflow’un bağlamını kullanarak varsayılan bir değer sağlayacağız. Bu, daha önce gördüğümüz, (piyasa kapalı olduğu için) yanıtın sınırlı bir anahtar kümesi içerdiği durumu çözer. Sonrasında listeyi bir pandas DataFrame’ine dönüştürüp döndüreceğiz. Ortaya çıkan görev şöyle görünür:

@task
def flatten_market_data(polygon_response, **context):
    # Create a list of headers and a list to store the normalized data in
    columns = {
        "status": "closed",
        "from": context.get("ds"),
        "symbol": "AMZN",
        "open": None,
        "high": None,
        "low": None,
        "close": None,
        "volume": None
    }
    # Create a list to append the data to
    flattened_record = []
    for header_name, default_value in columns.items():
        # Append the data
        flattened_record.append(polygon_response.get(header_name, default_value))
    # Convert to a pandas DataFrame
    flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
    return flattened_dataframeAltyapı:

hit_polygon_api ve flatten_market_data görevleri arasında bir bağımlılık eklememiz gerekecek. Bunu yapmak için DAG’ımızdaki kodu aşağıdakine uyacak şekilde güncelleyeceğiz:

import pandas as pd
with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    @task()
    def hit_polygon_api(**context):
        ...
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
# Set dependencies
    raw_market_data = hit_polygon_api()
    flatten_market_data(raw_market_data)Altyapı:

Burada, hit_polygon_api görevinin dönüş değeri raw_market_data değişkeninde saklanır. Ardından raw_market_data, polygon_response parametresi üzerinden flatten_market_data görevine argüman olarak geçirilir. Bu kod yalnızca hit_polygon_api ve flatten_market_data görevleri arasında bir bağımlılık oluşturmakla kalmaz, aynı zamanda bu iki görev arasında veri paylaşımına da olanak tanır.

Yaptığımız dönüşüm nispeten basit olsa da Airflow, çok daha karmaşık veri manipülasyonlarını yürütme olanağı sunar. Yerel görevlerin yanı sıra, AWS Lambda ve DBT gibi araçlarla dönüşümü orkestre etmek için Airflow’un kapsamlı sağlayıcı bağlantı kancaları ve operatör koleksiyonundan yararlanmak da kolaydır.

Airflow ile veri yükleme

ETL boru hattımızın son adımına geldik. Bunu bir SQLite veritabanı ve TaskFlow API ile tanımlanmış son bir görev kullanarak yapmayı planladık.

Daha önce olduğu gibi, görevimizi oluştururken tek bir parametre tanımlayacağız. Bunun adını flattened_dataframe koyacağız. Bu, flatten_market_data görevinin döndürdüğü verilerin yeni görevimize aktarılmasını sağlar.

Verileri bir SQLite veritabanına yükleyecek kodu yazmadan önce, Airflow arayüzünde bir bağlantı oluşturmamız gerekecek. Bağlantılar sayfasını açmak için şu adımları izleyin:

• Airflow arayüzünü açın
• Admin seçeneğinin üzerine gelin
• Connections öğesini seçin.
• Yeni bir bağlantı oluşturmak için + simgesine tıklayın.

Şuna benzeyen bir ekrana yönlendirileceksiniz:

Airflow bağlantıları sayfası

Bağlantılar sayfasını doldurmak için şu adımları izleyin:

• Connection Type’ı Sqlite olarak değiştirin.
• Connection Id için "market_database_conn" değerini girin.
• Host alanına "/usr/local/airflow/market_data.db" ekleyin.

Bu bağlantının yapılandırması aşağıdaki görsele benzemelidir. Bu hâle geldiğinde Save’e tıklayın.

SQLite veritabanına kaydedilmemiş Airflow bağlantısı

Artık bir bağlantı oluşturduğumuza göre, bu bilgiyi görevimizde SqliteHook kullanarak alabiliriz. Aşağıdaki koda bakın.

from airflow.providers.sqlite.hooks.sqlite import SqliteHookAltyapı:

@task
def load_market_data(flattened_dataframe):
    # Pull the connection
    market_database_hook = SqliteHook("market_database_conn")
    market_database_conn = market_database_hook.get_sqlalchemy_engine()
    # Load the table to Postgres, replace if it exists
    flattened_dataframe.to_sql(
        name="market_data",
        con=market_database_conn,
        if_exists="append",
        index=False
    )
    # print(market_database_hook.get_records("SELECT * FROM market_data;"))Altyapı:

Bu kodla, bir önceki adımda belirttiğimiz SQLite veritabanına bir bağlantı oluşturuyoruz. Ardından, .get_sqlalchemy_engine() yöntemi kullanılarak bağlantı motoru kancadan alınır. Bu, flattened_dataframe üzerinde .to_sql() yöntemi çağrılırken con parametresine argüman olarak aktarılır.

Bu verilerin yazıldığı tablo adının market_data olduğunu ve tablo mevcutsa ekleme yapıldığını unutmayın. Test ederken, kayıtları getirip yazdırarak verinin yazıldığını kontrol etmeyi severim. Bunu, bu görevin son satırının yorum satırını kaldırarak yapabilirsiniz.

Tüm bunları bir araya getirdiğimizde kodumuz aşağı yukarı şöyle görünmelidir:

from airflow import DAG
from airflow.decorators import task
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from datetime import datetime, timedelta
import requests
import pandas as pd

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
    @task
    def load_market_data(flattened_dataframe):
        # Pull the connection
        market_database_hook = SqliteHook("market_database_conn")
        market_database_conn = market_database_hook.get_sqlalchemy_engine()
        # Load the table to SQLite, append if it exists
        flattened_dataframe.to_sql(
            name="market_data",
            con=market_database_conn,
            if_exists="append",
            index=False
        )
    # Set dependencies between tasks
    raw_market_data = hit_polygon_api()
    transformed_market_data = flatten_market_data(raw_market_data)
    load_market_data(transformed_market_data)Altyapı:

Yine, flatten_market_data görevinden dönen verilerin load_market_data görevine aktarılmasına izin verecek şekilde bağımlılıklarımızı güncelledik. Ortaya çıkan DAG için grafik görünümü aşağıdakine benzer:

ETL boru hattı için grafik görünümü

Test

Artık ilk Airflow DAG’ınızı oluşturduğunuza göre, çalıştığından emin olma zamanı. Bunu yapmanın birkaç yolu vardır, ancak en yaygın olanlarından biri DAG’ı uçtan uca çalıştırmaktır.

Bunu Airflow arayüzünde yapmak için DAG’ınıza gidin ve anahtarı maviden aktif konuma getirin. catchup True olarak ayarlandığından, bir DAG çalıştırması kuyruğa alınmalı ve çalışmaya başlamalıdır. Bir görev başarıyla yürütülürse, arayüzde o göreve karşılık gelen kutu yeşile döner. DAG’deki tüm görevler başarılı olursa DAG success olarak işaretlenir ve bir sonraki DAG çalıştırması tetiklenir.

Bir görev başarısız olursa durum up for retry olur ve sarı ile işaretlenir. Bu olduğunda, en iyisi o görevin günlüklerine bakmaktır. Bunu, grid görünümünde sarı kutuya tıklayıp Logs’u seçerek yapabilirsiniz. Burada istisna mesajını bulacak ve triyaja başlayabileceksiniz. Bir görev belirtilen yeniden deneme sayısından fazla başarısız olursa, o görevin ve DAG’ın durumu failed olarak ayarlanır.

Bir DAG’ı uçtan uca test etmenin yanı sıra, Airflow birim testleri yazmayı kolaylaştırır. Ortamınızı ilk oluşturduğunuzda astro dev start ile bir tests/ dizini sizin için oluşturulur. Burada hem DAG’ınız hem de DAG’ınızın bileşenleri için birim testleri ekleyebilirsiniz.

Aşağıda DAG’ımızın yapılandırması için bir birim testi bulunmaktadır. Bu test, start_date, schedule ve catchup gibi tanımlandığında ayarlanan çeşitli parametreleri doğrular. Testinizi yazdıktan sonra projenizin kök dizinine gidin ve şunu çalıştırın:

from airflow.models.dagbag import DagBag
from datetime import datetime
import pytz

def test_market_etl_config():
    # Pull the DAG
    market_etl_dag = DagBag().get_dag("market_etl")
    # Assert start date, schedule, and catchup
    assert market_etl_dag.start_date == datetime(2024, 3, 25, 9, tzinfo=pytz.UTC)
    assert market_etl_dag.schedule_interval == "@daily"
    assert market_etl_dag.catchupAltyapı:

astro dev pytestAltyapı:

Bu komut, tests/ dizininizdeki tüm birim testlerini çalıştıracaktır. Yalnızca tek bir testi çalıştırmak isterseniz, yukarıdaki komuta dosyanın yolunu argüman olarak verebilirsiniz. Astro CLI’yı kullanmanın yanı sıra, herhangi bir Python test koşucu birim testleri yazmak ve yürütmek için kullanılabilir.

Kişisel projelerde birim testleri yazmak, kodunuzun beklediğiniz gibi çalışmasını sağlamanıza yardımcı olur. Kurumsal ortamlarda ise birim testleri neredeyse her zaman zorunludur. Çoğu veri ekibi Airflow projelerini dağıtmak için bir tür CI/CD aracı kullanır. Bu süreç genellikle birim testlerinin çalıştırılmasını ve çıktılarının doğrulanmasını içerir; böylece yazdığınız DAG’ın üretime hazır olduğundan emin olunur. Birim testleri hakkında daha fazla bilgi için Pytest ile Birim Testi Nasıl Yapılır öğreticimize ve Python’da Teste Giriş kursuna göz atın.

İleri Düzey Airflow İpuçları ve Teknikleri

Çalışan basit bir veri boru hattı kurduk ve dönüşüm ile kalıcılık tekniklerini de kullandık. Başka durumlarda Airflow, sağlayıcı yapımı ve özel operatörlerle terabaytlarca veriyi işleyen karmaşık iş akışlarını orkestre edecek donanıma sahiptir.

Bunlara örnek olarak, daha büyük bir veri yığınıyla kolayca entegrasyona olanak tanıyan S3ToSnowflakeOperator ve DatabricksRunNowOperator verilebilir. Bu tür operatörlerle hobisel bir ortamda çalışmak zordur. Örneğin, S3ToSnowflakeOperator’ü kullanmak için hem AWS hem de Snowflake hesaplarına ve veri aktaracağınız kaynaklar için yapılandırmaya ihtiyaç duyarsınız.

ETL iş akışlarına ek olarak, Airflow ELT iş akışlarını da destekler; bulut veri ambarlarını kullanan ekipler için endüstri standardı hâline gelmektedir. Veri boru hatları tasarlarken bunu aklınızda bulundurun.

Load bileşeninde, bir SQLite veritabanına bağlantı oluşturduk; bu bağlantı daha sonra alınarak verilerin kalıcı hâle getirilmesinde kullanıldı. Bazen Secrets olarak da adlandırılan bağlantılar, kaynak ve hedef sistemlerle etkileşimleri basitleştirmek için Airflow’da tasarlanmış bir özelliktir. Bu bağlantıları, örneğin Polygon API anahtarınız gibi hassas bilgileri saklamak için kullanarak kodunuzun güvenliğini artırırsınız. Bu yaklaşım, kimlik bilgilerini kod tabanınızdan ayrı yönetmenize de olanak tanır. Mümkün olduğunda, iş akışınızı hem güvenli hem de düzenli tutmak için Bağlantıları geniş ölçüde kullanmanız tavsiye edilir.

Dikkatinizi çekmiş olabilir: “AMZN” hisse sembolü hit_polygon_api ve flatten_market_data görevlerimize sabit kodlanmıştı. Bu, tek bir hisse sembolü için veri çıkarıp dönüştürmemize ve yüklememize olanak sağladı. Peki aynı kodu birden fazla hisse sembolü için kullanmak isterseniz? Neyse ki DAG’ları dinamik olarak üretmek kolaydır. Kodumuzda minimal bir yeniden düzenlemeyle bir hisse sembolleri listesi üzerinde döngü kurabilir ve sembol değerlerini parametreleştirebiliriz. Bu, DAG’larınızı daha modüler ve taşınabilir hâle getirir. DAG’ların dinamik olarak üretilmesi hakkında bilgi için Astronomer’ın Airflow’da DAG’ları Dinamik Olarak Üretme dokümantasyonuna bakın.

Sonuç

Tebrikler! Python, pandas ve SQLite kullanarak Polygon API’sinden borsa verilerini çıkarmak, dönüştürmek ve yüklemek için bir Airflow DAG’ı oluşturdunuz. Bu süreçte mimari diyagramlar ve teknik şartnameler hazırlama, Airflow bağlantıları oluşturma ve DAG’larınızı test etme konularında becerilerinizi geliştirdiniz. Airflow yolculuğunuza devam ederken, boru hatlarınızı sağlam, dayanıklı ve yeniden kullanılabilir kılacak daha gelişmiş tekniklerle denemeler yapın.  

Daha fazlasını öğrenmek için şu kaynaklara göz atın: 

• Python ile Airflow’a Giriş kursu
• Apache Airflow’a Başlarken
• Airflow vs Prefect karşılaştırması
• En iyi Airflow alternatifleri
• En iyi Airflow mülakat soruları