diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 00000000..4770cec6 Binary files /dev/null and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index e515a197..38c843a7 100644 --- a/.gitignore +++ b/.gitignore @@ -8,4 +8,7 @@ ecom/target/ ecom/dbt_packages/ ecom/logs/ ecom/.user.ymllogs/ -logs/ \ No newline at end of file +logs/airflow_venv/ +airflow/logs/ +airflow/airflow.db +airflow/__pycache__/ diff --git a/README.md b/README.md index a62f8387..c20e5ecf 100644 --- a/README.md +++ b/README.md @@ -4,28 +4,37 @@ ![Python](https://img.shields.io/badge/python-3.12-blue) ![dbt](https://img.shields.io/badge/dbt-1.11-orange) ![DuckDB](https://img.shields.io/badge/DuckDB-1.10-yellow) +![BigQuery](https://img.shields.io/badge/BigQuery-Google_Cloud-4285F4) +![Airflow](https://img.shields.io/badge/Airflow-2.9-017CEE) -A end-to-end analytics engineering project built on a **Trendyol-style e-commerce domain** — from raw synthetic data through a fully tested dbt star schema to a live Streamlit dashboard. Demonstrates modern data warehousing practices including SCD Type 2, CI/CD, and metric-layer design. +An end-to-end analytics engineering portfolio project built on a **Trendyol-style e-commerce domain**. Covers the full modern data stack: synthetic data generation, a fully tested dbt star schema, dual-engine deployment (DuckDB locally, BigQuery in production), Airflow orchestration, and a live Looker Studio dashboard. --- ## Architecture ``` -generate.py generate_changes.py - │ │ - ▼ ▼ -data/raw/ ──load_raw.py──▶ DuckDB (raw schema) - │ - dbt build - │ - ┌─────────────┼─────────────┐ - ▼ ▼ ▼ - stg_* views dim_* tables fact_orders - (staging) (marts) (marts) - │ - Streamlit - dashboard +generator/generate.py + │ + ▼ + data/raw/ (10 CSVs) + │ + ├──► scripts/load_raw.py ──► DuckDB (local dev) + │ │ + └──► scripts/load_bigquery.py ──► BigQuery raw dataset (production) + │ + dbt build + (--target bigquery) + │ + ┌────────────────┼────────────────┐ + ▼ ▼ ▼ + stg_* views dim_* tables fact_* tables + (staging) (marts) (marts) + │ + ┌──────────┴──────────┐ + ▼ ▼ + Airflow DAG Looker Studio + (orchestration) (dashboard) ``` --- @@ -35,58 +44,163 @@ data/raw/ ──load_raw.py──▶ DuckDB (raw schema) | Layer | Tool | |---|---| | Data generation | Python · Faker | -| Database | DuckDB | -| Transformation | dbt-duckdb | -| Testing | dbt schema tests | -| Dashboard | Streamlit · Plotly | +| Local database | DuckDB 1.10 | +| Production database | Google BigQuery | +| Transformation | dbt 1.11 (dbt-duckdb + dbt-bigquery) | +| Testing | dbt schema tests (100 tests) | +| Orchestration | Apache Airflow 2.9 | +| Dashboard | Looker Studio | | CI/CD | GitHub Actions | --- ## Data Model -### Star schema (`main_marts`) +### Star schema (19 models, 100/100 tests) ``` - dim_date - (182 rows) - │ - ┌───────────────┼───────────────┐ - │ │ │ - dim_customer fact_orders dim_product - dim_customer (4,531 rows) (400 rows) - _scd2 │ - (SCD Type 2) │ - dim_seller - (80 rows) + dim_date + (182 rows) + │ + ┌─────────────────────┼──────────────────────┐ + │ │ │ + dim_customer fact_orders dim_product + dim_customer_scd2 (4,500+ rows) (400 rows) + (SCD Type 2) │ + │ + ┌───────────┴───────────┐ + │ │ + dim_seller dim_campaign + dim_seller_scd2 (20 rows) + (SCD Type 2) + + fact_seller_daily + (incremental, 3,993 rows) ``` -**Fact table grain:** one row per order line item — the most granular level, allowing roll-up to order, customer, product, category, or day. +**Fact table grain:** one row per order line item — the most granular level, supporting roll-up to order, customer, product, category, campaign, or day. -### Staging layer (`main_staging`) +### Staging layer (10 models) | Model | Source | Key transformations | |---|---|---| -| `stg_customers` | `raw.customers` | Deduplicates re-registrations, removes soft-deletes | -| `stg_orders` | `raw.orders` | Parses two date formats (`ISO` + `dd/mm/yyyy`) via `TRY_STRPTIME` | -| `stg_order_items` | `raw.order_items` | Casts types, derives `line_revenue` and `is_flash_sale` | +| `stg_customers` | `raw.customers` | Deduplication, referential integrity | +| `stg_orders` | `raw.orders` | Dual date format parsing (`ISO` + `dd/mm/yyyy`), target-conditional SQL | +| `stg_order_items` | `raw.order_items` | Derives `line_revenue`, `is_flash_sale` | | `stg_products` | `raw.products` | Derives `gross_margin_pct` | | `stg_sellers` | `raw.sellers` | Type casting | -| `stg_sessions` | `raw.sessions` | Preserves anonymous sessions (null `customer_id`) | -| `stg_customer_changes` | `raw.customer_changes` | Cleans change events for SCD2 | +| `stg_sessions` | `raw.sessions` | Preserves anonymous sessions | +| `stg_customer_changes` | `raw.customer_changes` | SCD2 source events | +| `stg_seller_changes` | `raw.seller_changes` | Seller SCD2 source events | +| `stg_seller_daily_stats` | `raw.seller_daily_stats` | Aggregated seller metrics | +| `stg_campaigns` | `raw.campaigns` | Campaign metadata cleaning | -### SCD Type 2 — `dim_customer_scd2` +### SCD Type 2 — `dim_customer_scd2` and `dim_seller_scd2` -Tracks historical changes to customer `city` and `customer_tier`. Each version of a customer gets a separate row stamped with `valid_from` / `valid_to` dates, enabling point-in-time joins to the fact table. +Tracks historical changes to customer `city` / `customer_tier` and seller `rating` / `category`. Each version gets its own row with Trendyol's production column pattern: + +```sql +effective_date -- when this version became active +expiry_date -- when this version was superseded (NULL = current) +is_current_flag -- boolean shortcut for current version queries +``` + +Point-in-time join pattern: ```sql --- Get the customer's attributes at the time of their order JOIN dim_customer_scd2 scd - ON fact.customer_id = scd.customer_id - AND fact.order_date >= scd.valid_from - AND (fact.order_date < scd.valid_to OR scd.valid_to IS NULL) + ON fact.customer_id = scd.customer_id + AND fact.order_date >= scd.effective_date + AND (fact.order_date < scd.expiry_date OR scd.expiry_date IS NULL) +``` + +### Incremental models + +Both `fact_orders` and `fact_seller_daily` are incremental. On each run, only new records since the last watermark are inserted — no full table scans. + +BigQuery-compatible watermark pattern using CROSS JOIN to avoid subquery-in-WHERE restrictions: + +```sql +{% if is_incremental() %} +, watermark as ( + select coalesce(max(cast(order_date as date)), date '1900-01-01') as max_date + from {{ this }} +) +{% endif %} + +select joined.* +from joined +{% if is_incremental() %} +cross join watermark +where cast(joined.order_date as date) > watermark.max_date +{% endif %} +``` + +--- + +## Dual-Engine Deployment + +The same dbt models run against both DuckDB (local dev) and BigQuery (production) by switching a single line in `profiles.yml`: + +```yaml +ecom: + target: dev # ← change to 'bigquery' for production + outputs: + dev: + type: duckdb + path: "../data/ecom.duckdb" + bigquery: + type: bigquery + method: service-account + project: + dataset: marts + keyfile: ~/.dbt/dbt-ecom-key.json + location: EU +``` + +Engine-specific SQL is handled with `{% if target.type == 'bigquery' %}` blocks. Key differences resolved: + +| Pattern | DuckDB | BigQuery | +|---|---|---| +| Date parsing | `try_strptime()` | `safe.parse_date()` | +| Date spine | `generate_series()` | `generate_date_array()` | +| Date formatting | `strftime()` | `format_date()` | +| Decimal types | `decimal(10,2)` | `numeric` | +| Type casting | `::date` | `cast(x as date)` | + +--- + +## Airflow Orchestration + +Daily pipeline DAG (`ecom_daily_pipeline`) running at 06:00 UTC: + +``` +start + └── check_source_freshness (BranchPythonOperator) + ├── [FRESH] load_raw_to_bigquery + │ └── dbt_build_bigquery + │ └── notify_success + └── [STALE] notify_stale ``` +**Freshness check:** verifies `orders.csv` was modified within 24 hours before spending compute on a BigQuery build. If data is stale, the pipeline branches to `notify_stale` and skips the build entirely. + +**Dependency isolation:** Airflow runs in a dedicated venv (`airflow_venv/`) separate from the dbt venv (`.venv/`). This mirrors production deployments where orchestrators and the tools they call have independent dependency graphs. + +--- + +## Looker Studio Dashboard + +Live dashboard connected directly to BigQuery marts: + +🔗 [View Dashboard](https://datastudio.google.com/reporting/cd32c187-2909-4b3e-8456-19d2efd0b275) + +Charts: +- **Daily Revenue** — line chart over `order_date` +- **Revenue by Order Status** — bar chart (delivered, cancelled, processing, etc.) +- **Revenue by Platform** — pie chart (Web 49%, Android 40%, iOS 11%) +- **Total GMV** — scorecard + --- ## Project Structure @@ -95,28 +209,32 @@ JOIN dim_customer_scd2 scd e-commerce-analytics/ ├── .github/ │ └── workflows/ -│ └── dbt_ci.yml CI: full pipeline on every push + PR +│ └── dbt_ci.yml CI: full pipeline on every push + PR +├── airflow/ +│ └── dags/ +│ └── ecom_pipeline.py Airflow DAG (freshness → load → dbt → notify) ├── generator/ -│ └── generate.py Synthetic e-commerce data (Faker, seed=42) +│ └── generate.py Synthetic data generator (10 CSVs, Faker, seed=42) ├── scripts/ -│ ├── load_raw.py Loads CSVs into DuckDB raw schema -│ └── generate_changes.py Customer attribute change events (SCD2 source) -├── ecom/ dbt project +│ ├── load_raw.py Loads CSVs → DuckDB raw schema +│ └── load_bigquery.py Loads CSVs → BigQuery raw dataset +├── ecom/ dbt project │ ├── dbt_project.yml -│ ├── profiles.yml +│ ├── profiles.yml Dev (DuckDB) + bigquery targets │ └── models/ -│ ├── staging/ 7 stg_* views — clean, typed, tested -│ └── marts/ Star schema — 5 tables (4 dims + 1 fact) +│ ├── staging/ 10 stg_* views +│ └── marts/ 9 mart models │ ├── dim_date.sql │ ├── dim_customer.sql │ ├── dim_customer_scd2.sql -│ ├── dim_product.sql │ ├── dim_seller.sql -│ └── fact_orders.sql -├── dashboard/ -│ └── app.py Streamlit BI dashboard +│ ├── dim_seller_scd2.sql +│ ├── dim_product.sql +│ ├── dim_campaign.sql +│ ├── fact_orders.sql (incremental) +│ └── fact_seller_daily.sql (incremental) ├── data/ -│ └── raw/ Generated CSVs (git-ignored) +│ └── raw/ Generated CSVs (git-ignored) ├── requirements.txt └── .gitignore ``` @@ -125,95 +243,81 @@ e-commerce-analytics/ ## Quick Start -### 1. Clone and install +### Local (DuckDB) ```bash git clone https://github.com/b10000-coder/e-commerce-analytics.git cd e-commerce-analytics python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt -``` -### 2. Generate data +python generator/generate.py # generates 10 CSVs in data/raw/ +python scripts/load_raw.py # loads into DuckDB -```bash -python generator/generate.py # creates 6 CSVs in data/raw/ -python scripts/generate_changes.py # creates customer_changes.csv +cd ecom +dbt build --profiles-dir . # 100/100 tests ``` -### 3. Load into DuckDB +### Production (BigQuery) ```bash -python scripts/load_raw.py -``` - -Output: -``` -OK raw.customers 618 rows -OK raw.sellers 80 rows -OK raw.products 400 rows -OK raw.orders 2,500 rows -OK raw.order_items 4,881 rows -OK raw.sessions 4,000 rows -OK raw.customer_changes 142 rows -``` +# Authenticate GCP +gcloud auth application-default login -### 4. Run dbt (models + tests) +# Load raw data to BigQuery +python scripts/load_bigquery.py -```bash +# Run dbt against BigQuery cd ecom -dbt build --profiles-dir . +dbt build --profiles-dir . --target bigquery # 100/100 tests ``` -Expected output: -``` -Done. PASS=78 WARN=0 ERROR=0 SKIP=0 NO-OP=0 TOTAL=78 -``` - -### 5. Launch the dashboard +### Airflow ```bash -cd .. -streamlit run dashboard/app.py -``` +python -m venv airflow_venv && source airflow_venv/bin/activate +pip install "apache-airflow==2.9.1" \ + --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.9.1/constraints-3.12.txt" -Opens at `http://localhost:8501` +export AIRFLOW_HOME=$(pwd)/airflow +airflow standalone +# UI at http://localhost:8080 +``` --- ## dbt Test Coverage -78 tests across all models covering: +100 tests across all 19 models: - `unique` and `not_null` on all primary and surrogate keys -- `accepted_values` on status, tier, platform, and device_type columns -- `relationships` tests enforcing referential integrity across all foreign keys -- Point-in-time join integrity for SCD Type 2 +- `accepted_values` on `order_status`, `payment_method`, `customer_tier`, `campaign_type`, `device_type`, `platform` +- `relationships` enforcing referential integrity across all foreign keys +- Tests pass identically on both DuckDB and BigQuery --- ## CI/CD -Every push to `dev` and every PR to `main` triggers the full pipeline on a clean GitHub Actions runner: +Every push to `dev` and every PR to `main` triggers the full pipeline on GitHub Actions: ``` -checkout → install deps → generate data → load DuckDB → dbt build (78 tests) +checkout → install deps → generate data → load DuckDB → dbt build (100 tests) ``` -Total runtime: ~46 seconds. The compiled `ecom.duckdb` is uploaded as a workflow artifact. - --- ## Domain -Trendyol-style Turkish e-commerce — 6 months of data (Oct 2024 → Mar 2025): +Trendyol-style Turkish e-commerce — 6 months of synthetic data (Oct 2024 → Mar 2025): -- **8 product categories** — Electronics, Fashion, Home & Life, Beauty, Sports, Books, Toys -- **80 sellers** across major Turkish cities -- **~600 customers** with bronze / silver / gold tiers +- **8 product categories** — Electronics, Fashion, Home & Life, Beauty, Sports, Books, Toys, Food +- **80 sellers** across Turkish cities with rating and category change history +- **618 customers** with bronze / silver / gold tiers and city change history - **2,500 orders** / **4,881 line items** across iOS, Android, and web -- **4,000 sessions** including 20% anonymous (null customer_id) -- **Intentional data quality issues** — mixed date formats, duplicate customers, orphaned foreign keys — resolved in the staging layer +- **20 campaigns** — flash sale, coupon, seasonal, loyalty types +- **4,000 sessions** including anonymous (null `customer_id`) +- **Intentional data quality issues** — mixed date formats, orphaned foreign keys — resolved in the staging layer --- @@ -221,7 +325,9 @@ Trendyol-style Turkish e-commerce — 6 months of data (Oct 2024 → Mar 2025): - **Layered architecture** — raw → staging → marts with clear separation of concerns - **Star schema** — fact + dimension tables with surrogate keys -- **SCD Type 2** — full customer attribute history via `LEAD()` window function +- **SCD Type 2** — full customer and seller attribute history using `LEAD()` window functions and Trendyol's `effective_date / expiry_date / is_current_flag` pattern +- **Incremental models** — watermark-based append-only loads on both fact tables +- **Dual-engine SQL** — target-conditional Jinja blocks, same models on DuckDB and BigQuery +- **Orchestration** — Airflow DAG with branching, freshness gates, and retry logic - **dbt best practices** — sources, refs, schema tests, documented models -- **CI/CD for data** — reproducible pipeline from raw data to tested marts in one command -- **Metric layer** — GMV, AOV, gross margin, flash sale rate defined once in SQL +- **Production deployment** — GCP service account auth, BigQuery datasets, Looker Studio BI layer diff --git a/airflow/airflow-webserver.pid b/airflow/airflow-webserver.pid new file mode 100644 index 00000000..b970ea89 --- /dev/null +++ b/airflow/airflow-webserver.pid @@ -0,0 +1 @@ +43853 diff --git a/airflow/airflow.cfg b/airflow/airflow.cfg new file mode 100644 index 00000000..a84b42ab --- /dev/null +++ b/airflow/airflow.cfg @@ -0,0 +1,2369 @@ +[core] +# The folder where your airflow pipelines live, most likely a +# subfolder in a code repository. This path must be absolute. +# +# Variable: AIRFLOW__CORE__DAGS_FOLDER +# +dags_folder = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/dags + +# Hostname by providing a path to a callable, which will resolve the hostname. +# The format is "package.function". +# +# For example, default value "airflow.utils.net.getfqdn" means that result from patched +# version of socket.getfqdn() - see https://github.com/python/cpython/issues/49254. +# +# No argument should be required in the function specified. +# If using IP address as hostname is preferred, use value ``airflow.utils.net.get_host_ip_address`` +# +# Variable: AIRFLOW__CORE__HOSTNAME_CALLABLE +# +hostname_callable = airflow.utils.net.getfqdn + +# A callable to check if a python file has airflow dags defined or not +# with argument as: `(file_path: str, zip_file: zipfile.ZipFile | None = None)` +# return True if it has dags otherwise False +# If this is not provided, Airflow uses its own heuristic rules. +# +# Variable: AIRFLOW__CORE__MIGHT_CONTAIN_DAG_CALLABLE +# +might_contain_dag_callable = airflow.utils.file.might_contain_dag_via_default_heuristic + +# Default timezone in case supplied date times are naive +# can be utc (default), system, or any IANA timezone string (e.g. Europe/Amsterdam) +# +# Variable: AIRFLOW__CORE__DEFAULT_TIMEZONE +# +default_timezone = utc + +# The executor class that airflow should use. Choices include +# ``SequentialExecutor``, ``LocalExecutor``, ``CeleryExecutor``, +# ``KubernetesExecutor``, ``CeleryKubernetesExecutor``, ``LocalKubernetesExecutor`` or the +# full import path to the class when using a custom executor. +# +# Variable: AIRFLOW__CORE__EXECUTOR +# +executor = SequentialExecutor + +# The auth manager class that airflow should use. Full import path to the auth manager class. +# +# Variable: AIRFLOW__CORE__AUTH_MANAGER +# +auth_manager = airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager + +# This defines the maximum number of task instances that can run concurrently per scheduler in +# Airflow, regardless of the worker count. Generally this value, multiplied by the number of +# schedulers in your cluster, is the maximum number of task instances with the running +# state in the metadata database. +# +# Variable: AIRFLOW__CORE__PARALLELISM +# +parallelism = 32 + +# The maximum number of task instances allowed to run concurrently in each DAG. To calculate +# the number of tasks that is running concurrently for a DAG, add up the number of running +# tasks for all DAG runs of the DAG. This is configurable at the DAG level with ``max_active_tasks``, +# which is defaulted as ``max_active_tasks_per_dag``. +# +# An example scenario when this would be useful is when you want to stop a new dag with an early +# start date from stealing all the executor slots in a cluster. +# +# Variable: AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG +# +max_active_tasks_per_dag = 16 + +# Are DAGs paused by default at creation +# +# Variable: AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION +# +dags_are_paused_at_creation = True + +# The maximum number of active DAG runs per DAG. The scheduler will not create more DAG runs +# if it reaches the limit. This is configurable at the DAG level with ``max_active_runs``, +# which is defaulted as ``max_active_runs_per_dag``. +# +# Variable: AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG +# +max_active_runs_per_dag = 16 + +# (experimental) The maximum number of consecutive DAG failures before DAG is automatically paused. +# This is also configurable per DAG level with ``max_consecutive_failed_dag_runs``, +# which is defaulted as ``max_consecutive_failed_dag_runs_per_dag``. +# If not specified, then the value is considered as 0, +# meaning that the dags are never paused out by default. +# +# Variable: AIRFLOW__CORE__MAX_CONSECUTIVE_FAILED_DAG_RUNS_PER_DAG +# +max_consecutive_failed_dag_runs_per_dag = 0 + +# The name of the method used in order to start Python processes via the multiprocessing module. +# This corresponds directly with the options available in the Python docs: +# https://docs.python.org/3/library/multiprocessing.html#multiprocessing.set_start_method. +# Must be one of the values returned by: +# https://docs.python.org/3/library/multiprocessing.html#multiprocessing.get_all_start_methods. +# +# Example: mp_start_method = fork +# +# Variable: AIRFLOW__CORE__MP_START_METHOD +# +# mp_start_method = + +# Whether to load the DAG examples that ship with Airflow. It's good to +# get started, but you probably want to set this to ``False`` in a production +# environment +# +# Variable: AIRFLOW__CORE__LOAD_EXAMPLES +# +load_examples = True + +# Path to the folder containing Airflow plugins +# +# Variable: AIRFLOW__CORE__PLUGINS_FOLDER +# +plugins_folder = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/plugins + +# Should tasks be executed via forking of the parent process ("False", +# the speedier option) or by spawning a new python process ("True" slow, +# but means plugin changes picked up by tasks straight away) +# +# Variable: AIRFLOW__CORE__EXECUTE_TASKS_NEW_PYTHON_INTERPRETER +# +execute_tasks_new_python_interpreter = False + +# Secret key to save connection passwords in the db +# +# Variable: AIRFLOW__CORE__FERNET_KEY +# +fernet_key = + +# Whether to disable pickling dags +# +# Variable: AIRFLOW__CORE__DONOT_PICKLE +# +donot_pickle = True + +# How long before timing out a python file import +# +# Variable: AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT +# +dagbag_import_timeout = 30.0 + +# Should a traceback be shown in the UI for dagbag import errors, +# instead of just the exception message +# +# Variable: AIRFLOW__CORE__DAGBAG_IMPORT_ERROR_TRACEBACKS +# +dagbag_import_error_tracebacks = True + +# If tracebacks are shown, how many entries from the traceback should be shown +# +# Variable: AIRFLOW__CORE__DAGBAG_IMPORT_ERROR_TRACEBACK_DEPTH +# +dagbag_import_error_traceback_depth = 2 + +# How long before timing out a DagFileProcessor, which processes a dag file +# +# Variable: AIRFLOW__CORE__DAG_FILE_PROCESSOR_TIMEOUT +# +dag_file_processor_timeout = 50 + +# The class to use for running task instances in a subprocess. +# Choices include StandardTaskRunner, CgroupTaskRunner or the full import path to the class +# when using a custom task runner. +# +# Variable: AIRFLOW__CORE__TASK_RUNNER +# +task_runner = StandardTaskRunner + +# If set, tasks without a ``run_as_user`` argument will be run with this user +# Can be used to de-elevate a sudo user running Airflow when executing tasks +# +# Variable: AIRFLOW__CORE__DEFAULT_IMPERSONATION +# +default_impersonation = + +# What security module to use (for example kerberos) +# +# Variable: AIRFLOW__CORE__SECURITY +# +security = + +# Turn unit test mode on (overwrites many configuration options with test +# values at runtime) +# +# Variable: AIRFLOW__CORE__UNIT_TEST_MODE +# +unit_test_mode = False + +# Whether to enable pickling for xcom (note that this is insecure and allows for +# RCE exploits). +# +# Variable: AIRFLOW__CORE__ENABLE_XCOM_PICKLING +# +enable_xcom_pickling = False + +# What classes can be imported during deserialization. This is a multi line value. +# The individual items will be parsed as a pattern to a glob function. +# Python built-in classes (like dict) are always allowed. +# +# Variable: AIRFLOW__CORE__ALLOWED_DESERIALIZATION_CLASSES +# +allowed_deserialization_classes = airflow.* + +# What classes can be imported during deserialization. This is a multi line value. +# The individual items will be parsed as regexp patterns. +# This is a secondary option to ``allowed_deserialization_classes``. +# +# Variable: AIRFLOW__CORE__ALLOWED_DESERIALIZATION_CLASSES_REGEXP +# +allowed_deserialization_classes_regexp = + +# When a task is killed forcefully, this is the amount of time in seconds that +# it has to cleanup after it is sent a SIGTERM, before it is SIGKILLED +# +# Variable: AIRFLOW__CORE__KILLED_TASK_CLEANUP_TIME +# +killed_task_cleanup_time = 60 + +# Whether to override params with dag_run.conf. If you pass some key-value pairs +# through ``airflow dags backfill -c`` or +# ``airflow dags trigger -c``, the key-value pairs will override the existing ones in params. +# +# Variable: AIRFLOW__CORE__DAG_RUN_CONF_OVERRIDES_PARAMS +# +dag_run_conf_overrides_params = True + +# If enabled, Airflow will only scan files containing both ``DAG`` and ``airflow`` (case-insensitive). +# +# Variable: AIRFLOW__CORE__DAG_DISCOVERY_SAFE_MODE +# +dag_discovery_safe_mode = True + +# The pattern syntax used in the ".airflowignore" files in the DAG directories. Valid values are +# ``regexp`` or ``glob``. +# +# Variable: AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX +# +dag_ignore_file_syntax = regexp + +# The number of retries each task is going to have by default. Can be overridden at dag or task level. +# +# Variable: AIRFLOW__CORE__DEFAULT_TASK_RETRIES +# +default_task_retries = 0 + +# The number of seconds each task is going to wait by default between retries. Can be overridden at +# dag or task level. +# +# Variable: AIRFLOW__CORE__DEFAULT_TASK_RETRY_DELAY +# +default_task_retry_delay = 300 + +# The maximum delay (in seconds) each task is going to wait by default between retries. +# This is a global setting and cannot be overridden at task or DAG level. +# +# Variable: AIRFLOW__CORE__MAX_TASK_RETRY_DELAY +# +max_task_retry_delay = 86400 + +# The weighting method used for the effective total priority weight of the task +# +# Variable: AIRFLOW__CORE__DEFAULT_TASK_WEIGHT_RULE +# +default_task_weight_rule = downstream + +# The default task execution_timeout value for the operators. Expected an integer value to +# be passed into timedelta as seconds. If not specified, then the value is considered as None, +# meaning that the operators are never timed out by default. +# +# Variable: AIRFLOW__CORE__DEFAULT_TASK_EXECUTION_TIMEOUT +# +default_task_execution_timeout = + +# Updating serialized DAG can not be faster than a minimum interval to reduce database write rate. +# +# Variable: AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL +# +min_serialized_dag_update_interval = 30 + +# If True, serialized DAGs are compressed before writing to DB. +# Note: this will disable the DAG dependencies view +# +# Variable: AIRFLOW__CORE__COMPRESS_SERIALIZED_DAGS +# +compress_serialized_dags = False + +# Fetching serialized DAG can not be faster than a minimum interval to reduce database +# read rate. This config controls when your DAGs are updated in the Webserver +# +# Variable: AIRFLOW__CORE__MIN_SERIALIZED_DAG_FETCH_INTERVAL +# +min_serialized_dag_fetch_interval = 10 + +# Maximum number of Rendered Task Instance Fields (Template Fields) per task to store +# in the Database. +# All the template_fields for each of Task Instance are stored in the Database. +# Keeping this number small may cause an error when you try to view ``Rendered`` tab in +# TaskInstance view for older tasks. +# +# Variable: AIRFLOW__CORE__MAX_NUM_RENDERED_TI_FIELDS_PER_TASK +# +max_num_rendered_ti_fields_per_task = 30 + +# On each dagrun check against defined SLAs +# +# Variable: AIRFLOW__CORE__CHECK_SLAS +# +check_slas = True + +# Path to custom XCom class that will be used to store and resolve operators results +# +# Example: xcom_backend = path.to.CustomXCom +# +# Variable: AIRFLOW__CORE__XCOM_BACKEND +# +xcom_backend = airflow.models.xcom.BaseXCom + +# By default Airflow plugins are lazily-loaded (only loaded when required). Set it to ``False``, +# if you want to load plugins whenever 'airflow' is invoked via cli or loaded from module. +# +# Variable: AIRFLOW__CORE__LAZY_LOAD_PLUGINS +# +lazy_load_plugins = True + +# By default Airflow providers are lazily-discovered (discovery and imports happen only when required). +# Set it to False, if you want to discover providers whenever 'airflow' is invoked via cli or +# loaded from module. +# +# Variable: AIRFLOW__CORE__LAZY_DISCOVER_PROVIDERS +# +lazy_discover_providers = True + +# Hide sensitive Variables or Connection extra json keys from UI and task logs when set to True +# +# (Connection passwords are always hidden in logs) +# +# Variable: AIRFLOW__CORE__HIDE_SENSITIVE_VAR_CONN_FIELDS +# +hide_sensitive_var_conn_fields = True + +# A comma-separated list of extra sensitive keywords to look for in variables names or connection's +# extra JSON. +# +# Variable: AIRFLOW__CORE__SENSITIVE_VAR_CONN_NAMES +# +sensitive_var_conn_names = + +# Task Slot counts for ``default_pool``. This setting would not have any effect in an existing +# deployment where the ``default_pool`` is already created. For existing deployments, users can +# change the number of slots using Webserver, API or the CLI +# +# Variable: AIRFLOW__CORE__DEFAULT_POOL_TASK_SLOT_COUNT +# +default_pool_task_slot_count = 128 + +# The maximum list/dict length an XCom can push to trigger task mapping. If the pushed list/dict has a +# length exceeding this value, the task pushing the XCom will be failed automatically to prevent the +# mapped tasks from clogging the scheduler. +# +# Variable: AIRFLOW__CORE__MAX_MAP_LENGTH +# +max_map_length = 1024 + +# The default umask to use for process when run in daemon mode (scheduler, worker, etc.) +# +# This controls the file-creation mode mask which determines the initial value of file permission bits +# for newly created files. +# +# This value is treated as an octal-integer. +# +# Variable: AIRFLOW__CORE__DAEMON_UMASK +# +daemon_umask = 0o077 + +# Class to use as dataset manager. +# +# Example: dataset_manager_class = airflow.datasets.manager.DatasetManager +# +# Variable: AIRFLOW__CORE__DATASET_MANAGER_CLASS +# +# dataset_manager_class = + +# Kwargs to supply to dataset manager. +# +# Example: dataset_manager_kwargs = {"some_param": "some_value"} +# +# Variable: AIRFLOW__CORE__DATASET_MANAGER_KWARGS +# +# dataset_manager_kwargs = + +# (experimental) Whether components should use Airflow Internal API for DB connectivity. +# +# Variable: AIRFLOW__CORE__DATABASE_ACCESS_ISOLATION +# +database_access_isolation = False + +# (experimental) Airflow Internal API url. Only used if [core] database_access_isolation is True. +# +# Example: internal_api_url = http://localhost:8080 +# +# Variable: AIRFLOW__CORE__INTERNAL_API_URL +# +# internal_api_url = + +# The ability to allow testing connections across Airflow UI, API and CLI. +# Supported options: Disabled, Enabled, Hidden. Default: Disabled +# Disabled - Disables the test connection functionality and disables the Test Connection button in UI. +# Enabled - Enables the test connection functionality and shows the Test Connection button in UI. +# Hidden - Disables the test connection functionality and hides the Test Connection button in UI. +# Before setting this to Enabled, make sure that you review the users who are able to add/edit +# connections and ensure they are trusted. Connection testing can be done maliciously leading to +# undesired and insecure outcomes. For more information on capabilities of users, see the documentation: +# https://airflow.apache.org/docs/apache-airflow/stable/security/security_model.html#capabilities-of-authenticated-ui-users +# +# Variable: AIRFLOW__CORE__TEST_CONNECTION +# +test_connection = Disabled + +# The maximum length of the rendered template field. If the value to be stored in the +# rendered template field exceeds this size, it's redacted. +# +# Variable: AIRFLOW__CORE__MAX_TEMPLATED_FIELD_LENGTH +# +max_templated_field_length = 4096 + +[database] +# Path to the ``alembic.ini`` file. You can either provide the file path relative +# to the Airflow home directory or the absolute path if it is located elsewhere. +# +# Variable: AIRFLOW__DATABASE__ALEMBIC_INI_FILE_PATH +# +alembic_ini_file_path = alembic.ini + +# The SqlAlchemy connection string to the metadata database. +# SqlAlchemy supports many different database engines. +# More information here: +# http://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#database-uri +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN +# +sql_alchemy_conn = sqlite:////Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/airflow.db + +# Extra engine specific keyword args passed to SQLAlchemy's create_engine, as a JSON-encoded value +# +# Example: sql_alchemy_engine_args = {"arg1": True} +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_ENGINE_ARGS +# +# sql_alchemy_engine_args = + +# The encoding for the databases +# +# Variable: AIRFLOW__DATABASE__SQL_ENGINE_ENCODING +# +sql_engine_encoding = utf-8 + +# Collation for ``dag_id``, ``task_id``, ``key``, ``external_executor_id`` columns +# in case they have different encoding. +# By default this collation is the same as the database collation, however for ``mysql`` and ``mariadb`` +# the default is ``utf8mb3_bin`` so that the index sizes of our index keys will not exceed +# the maximum size of allowed index when collation is set to ``utf8mb4`` variant +# (see https://github.com/apache/airflow/pull/17603#issuecomment-901121618). +# +# Variable: AIRFLOW__DATABASE__SQL_ENGINE_COLLATION_FOR_IDS +# +# sql_engine_collation_for_ids = + +# If SqlAlchemy should pool database connections. +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_ENABLED +# +sql_alchemy_pool_enabled = True + +# The SqlAlchemy pool size is the maximum number of database connections +# in the pool. 0 indicates no limit. +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_SIZE +# +sql_alchemy_pool_size = 5 + +# The maximum overflow size of the pool. +# When the number of checked-out connections reaches the size set in pool_size, +# additional connections will be returned up to this limit. +# When those additional connections are returned to the pool, they are disconnected and discarded. +# It follows then that the total number of simultaneous connections the pool will allow +# is pool_size + max_overflow, +# and the total number of "sleeping" connections the pool will allow is pool_size. +# max_overflow can be set to ``-1`` to indicate no overflow limit; +# no limit will be placed on the total number of concurrent connections. Defaults to ``10``. +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_OVERFLOW +# +sql_alchemy_max_overflow = 10 + +# The SqlAlchemy pool recycle is the number of seconds a connection +# can be idle in the pool before it is invalidated. This config does +# not apply to sqlite. If the number of DB connections is ever exceeded, +# a lower config value will allow the system to recover faster. +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_RECYCLE +# +sql_alchemy_pool_recycle = 1800 + +# Check connection at the start of each connection pool checkout. +# Typically, this is a simple statement like "SELECT 1". +# More information here: +# https://docs.sqlalchemy.org/en/14/core/pooling.html#disconnect-handling-pessimistic +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_PRE_PING +# +sql_alchemy_pool_pre_ping = True + +# The schema to use for the metadata database. +# SqlAlchemy supports databases with the concept of multiple schemas. +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_SCHEMA +# +sql_alchemy_schema = + +# Import path for connect args in SqlAlchemy. Defaults to an empty dict. +# This is useful when you want to configure db engine args that SqlAlchemy won't parse +# in connection string. This can be set by passing a dictionary containing the create engine parameters. +# For more details about passing create engine parameters(keepalives variables, timeout etc) +# in postgres db connection see +# https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#setting-up-a-postgresql-database +# e.g connect_args={"timeout":30} can be defined in airflow_local_settings.py and +# can be imported as shown below +# +# Example: sql_alchemy_connect_args = airflow_local_settings.connect_args +# +# Variable: AIRFLOW__DATABASE__SQL_ALCHEMY_CONNECT_ARGS +# +# sql_alchemy_connect_args = + +# Whether to load the default connections that ship with Airflow when ``airflow db init`` is called. +# It's good to get started, but you probably want to set this to ``False`` in a production environment. +# +# Variable: AIRFLOW__DATABASE__LOAD_DEFAULT_CONNECTIONS +# +load_default_connections = True + +# Number of times the code should be retried in case of DB Operational Errors. +# Not all transactions will be retried as it can cause undesired state. +# Currently it is only used in ``DagFileProcessor.process_file`` to retry ``dagbag.sync_to_db``. +# +# Variable: AIRFLOW__DATABASE__MAX_DB_RETRIES +# +max_db_retries = 3 + +# Whether to run alembic migrations during Airflow start up. Sometimes this operation can be expensive, +# and the users can assert the correct version through other means (e.g. through a Helm chart). +# Accepts "True" or "False". +# +# Variable: AIRFLOW__DATABASE__CHECK_MIGRATIONS +# +check_migrations = True + +[logging] +# The folder where airflow should store its log files. +# This path must be absolute. +# There are a few existing configurations that assume this is set to the default. +# If you choose to override this you may need to update the dag_processor_manager_log_location and +# child_process_log_directory settings as well. +# +# Variable: AIRFLOW__LOGGING__BASE_LOG_FOLDER +# +base_log_folder = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/logs + +# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search. +# Set this to True if you want to enable remote logging. +# +# Variable: AIRFLOW__LOGGING__REMOTE_LOGGING +# +remote_logging = False + +# Users must supply an Airflow connection id that provides access to the storage +# location. Depending on your remote logging service, this may only be used for +# reading logs, not writing them. +# +# Variable: AIRFLOW__LOGGING__REMOTE_LOG_CONN_ID +# +remote_log_conn_id = + +# Whether the local log files for GCS, S3, WASB and OSS remote logging should be deleted after +# they are uploaded to the remote location. +# +# Variable: AIRFLOW__LOGGING__DELETE_LOCAL_LOGS +# +delete_local_logs = False + +# Path to Google Credential JSON file. If omitted, authorization based on `the Application Default +# Credentials +# `__ will +# be used. +# +# Variable: AIRFLOW__LOGGING__GOOGLE_KEY_PATH +# +google_key_path = + +# Storage bucket URL for remote logging +# S3 buckets should start with "s3://" +# Cloudwatch log groups should start with "cloudwatch://" +# GCS buckets should start with "gs://" +# WASB buckets should start with "wasb" just to help Airflow select correct handler +# Stackdriver logs should start with "stackdriver://" +# +# Variable: AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER +# +remote_base_log_folder = + +# The remote_task_handler_kwargs param is loaded into a dictionary and passed to __init__ of remote +# task handler and it overrides the values provided by Airflow config. For example if you set +# `delete_local_logs=False` and you provide ``{"delete_local_copy": true}``, then the local +# log files will be deleted after they are uploaded to remote location. +# +# Example: remote_task_handler_kwargs = {"delete_local_copy": true} +# +# Variable: AIRFLOW__LOGGING__REMOTE_TASK_HANDLER_KWARGS +# +remote_task_handler_kwargs = + +# Use server-side encryption for logs stored in S3 +# +# Variable: AIRFLOW__LOGGING__ENCRYPT_S3_LOGS +# +encrypt_s3_logs = False + +# Logging level. +# +# Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``. +# +# Variable: AIRFLOW__LOGGING__LOGGING_LEVEL +# +logging_level = INFO + +# Logging level for celery. If not set, it uses the value of logging_level +# +# Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``. +# +# Variable: AIRFLOW__LOGGING__CELERY_LOGGING_LEVEL +# +celery_logging_level = + +# Logging level for Flask-appbuilder UI. +# +# Supported values: ``CRITICAL``, ``ERROR``, ``WARNING``, ``INFO``, ``DEBUG``. +# +# Variable: AIRFLOW__LOGGING__FAB_LOGGING_LEVEL +# +fab_logging_level = WARNING + +# Logging class +# Specify the class that will specify the logging configuration +# This class has to be on the python classpath +# +# Example: logging_config_class = my.path.default_local_settings.LOGGING_CONFIG +# +# Variable: AIRFLOW__LOGGING__LOGGING_CONFIG_CLASS +# +logging_config_class = + +# Flag to enable/disable Colored logs in Console +# Colour the logs when the controlling terminal is a TTY. +# +# Variable: AIRFLOW__LOGGING__COLORED_CONSOLE_LOG +# +colored_console_log = True + +# Log format for when Colored logs is enabled +# +# Variable: AIRFLOW__LOGGING__COLORED_LOG_FORMAT +# +colored_log_format = [%%(blue)s%%(asctime)s%%(reset)s] {%%(blue)s%%(filename)s:%%(reset)s%%(lineno)d} %%(log_color)s%%(levelname)s%%(reset)s - %%(log_color)s%%(message)s%%(reset)s + +# Specifies the class utilized by Airflow to implement colored logging +# +# Variable: AIRFLOW__LOGGING__COLORED_FORMATTER_CLASS +# +colored_formatter_class = airflow.utils.log.colored_log.CustomTTYColoredFormatter + +# Format of Log line +# +# Variable: AIRFLOW__LOGGING__LOG_FORMAT +# +log_format = [%%(asctime)s] {%%(filename)s:%%(lineno)d} %%(levelname)s - %%(message)s + +# Defines the format of log messages for simple logging configuration +# +# Variable: AIRFLOW__LOGGING__SIMPLE_LOG_FORMAT +# +simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s + +# Where to send dag parser logs. If "file", logs are sent to log files defined by child_process_log_directory. +# +# Variable: AIRFLOW__LOGGING__DAG_PROCESSOR_LOG_TARGET +# +dag_processor_log_target = file + +# Format of Dag Processor Log line +# +# Variable: AIRFLOW__LOGGING__DAG_PROCESSOR_LOG_FORMAT +# +dag_processor_log_format = [%%(asctime)s] [SOURCE:DAG_PROCESSOR] {%%(filename)s:%%(lineno)d} %%(levelname)s - %%(message)s + +# Determines the formatter class used by Airflow for structuring its log messages +# The default formatter class is timezone-aware, which means that timestamps attached to log entries +# will be adjusted to reflect the local timezone of the Airflow instance +# +# Variable: AIRFLOW__LOGGING__LOG_FORMATTER_CLASS +# +log_formatter_class = airflow.utils.log.timezone_aware.TimezoneAware + +# An import path to a function to add adaptations of each secret added with +# `airflow.utils.log.secrets_masker.mask_secret` to be masked in log messages. The given function +# is expected to require a single parameter: the secret to be adapted. It may return a +# single adaptation of the secret or an iterable of adaptations to each be masked as secrets. +# The original secret will be masked as well as any adaptations returned. +# +# Example: secret_mask_adapter = urllib.parse.quote +# +# Variable: AIRFLOW__LOGGING__SECRET_MASK_ADAPTER +# +secret_mask_adapter = + +# Specify prefix pattern like mentioned below with stream handler TaskHandlerWithCustomFormatter +# +# Example: task_log_prefix_template = {ti.dag_id}-{ti.task_id}-{execution_date}-{try_number} +# +# Variable: AIRFLOW__LOGGING__TASK_LOG_PREFIX_TEMPLATE +# +task_log_prefix_template = + +# Formatting for how airflow generates file names/paths for each task run. +# +# Variable: AIRFLOW__LOGGING__LOG_FILENAME_TEMPLATE +# +log_filename_template = dag_id={{ ti.dag_id }}/run_id={{ ti.run_id }}/task_id={{ ti.task_id }}/{%% if ti.map_index >= 0 %%}map_index={{ ti.map_index }}/{%% endif %%}attempt={{ try_number }}.log + +# Formatting for how airflow generates file names for log +# +# Variable: AIRFLOW__LOGGING__LOG_PROCESSOR_FILENAME_TEMPLATE +# +log_processor_filename_template = {{ filename }}.log + +# Full path of dag_processor_manager logfile. +# +# Variable: AIRFLOW__LOGGING__DAG_PROCESSOR_MANAGER_LOG_LOCATION +# +dag_processor_manager_log_location = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/logs/dag_processor_manager/dag_processor_manager.log + +# Whether DAG processor manager will write logs to stdout +# +# Variable: AIRFLOW__LOGGING__DAG_PROCESSOR_MANAGER_LOG_STDOUT +# +dag_processor_manager_log_stdout = False + +# Name of handler to read task instance logs. +# Defaults to use ``task`` handler. +# +# Variable: AIRFLOW__LOGGING__TASK_LOG_READER +# +task_log_reader = task + +# A comma\-separated list of third-party logger names that will be configured to print messages to +# consoles\. +# +# Example: extra_logger_names = connexion,sqlalchemy +# +# Variable: AIRFLOW__LOGGING__EXTRA_LOGGER_NAMES +# +extra_logger_names = + +# When you start an airflow worker, airflow starts a tiny web server +# subprocess to serve the workers local log files to the airflow main +# web server, who then builds pages and sends them to users. This defines +# the port on which the logs are served. It needs to be unused, and open +# visible from the main web server to connect into the workers. +# +# Variable: AIRFLOW__LOGGING__WORKER_LOG_SERVER_PORT +# +worker_log_server_port = 8793 + +# Port to serve logs from for triggerer. See worker_log_server_port description +# for more info. +# +# Variable: AIRFLOW__LOGGING__TRIGGER_LOG_SERVER_PORT +# +trigger_log_server_port = 8794 + +# We must parse timestamps to interleave logs between trigger and task. To do so, +# we need to parse timestamps in log files. In case your log format is non-standard, +# you may provide import path to callable which takes a string log line and returns +# the timestamp (datetime.datetime compatible). +# +# Example: interleave_timestamp_parser = path.to.my_func +# +# Variable: AIRFLOW__LOGGING__INTERLEAVE_TIMESTAMP_PARSER +# +# interleave_timestamp_parser = + +# Permissions in the form or of octal string as understood by chmod. The permissions are important +# when you use impersonation, when logs are written by a different user than airflow. The most secure +# way of configuring it in this case is to add both users to the same group and make it the default +# group of both users. Group-writeable logs are default in airflow, but you might decide that you are +# OK with having the logs other-writeable, in which case you should set it to `0o777`. You might +# decide to add more security if you do not use impersonation and change it to `0o755` to make it +# only owner-writeable. You can also make it just readable only for owner by changing it to `0o700` if +# all the access (read/write) for your logs happens from the same user. +# +# Example: file_task_handler_new_folder_permissions = 0o775 +# +# Variable: AIRFLOW__LOGGING__FILE_TASK_HANDLER_NEW_FOLDER_PERMISSIONS +# +file_task_handler_new_folder_permissions = 0o775 + +# Permissions in the form or of octal string as understood by chmod. The permissions are important +# when you use impersonation, when logs are written by a different user than airflow. The most secure +# way of configuring it in this case is to add both users to the same group and make it the default +# group of both users. Group-writeable logs are default in airflow, but you might decide that you are +# OK with having the logs other-writeable, in which case you should set it to `0o666`. You might +# decide to add more security if you do not use impersonation and change it to `0o644` to make it +# only owner-writeable. You can also make it just readable only for owner by changing it to `0o600` if +# all the access (read/write) for your logs happens from the same user. +# +# Example: file_task_handler_new_file_permissions = 0o664 +# +# Variable: AIRFLOW__LOGGING__FILE_TASK_HANDLER_NEW_FILE_PERMISSIONS +# +file_task_handler_new_file_permissions = 0o664 + +# By default Celery sends all logs into stderr. +# If enabled any previous logging handlers will get *removed*. +# With this option AirFlow will create new handlers +# and send low level logs like INFO and WARNING to stdout, +# while sending higher severity logs to stderr. +# +# Variable: AIRFLOW__LOGGING__CELERY_STDOUT_STDERR_SEPARATION +# +celery_stdout_stderr_separation = False + +# If enabled, Airflow may ship messages to task logs from outside the task run context, e.g. from +# the scheduler, executor, or callback execution context. This can help in circumstances such as +# when there's something blocking the execution of the task and ordinarily there may be no task +# logs at all. +# This is set to True by default. If you encounter issues with this feature +# (e.g. scheduler performance issues) it can be disabled. +# +# Variable: AIRFLOW__LOGGING__ENABLE_TASK_CONTEXT_LOGGER +# +enable_task_context_logger = True + +[metrics] +# StatsD (https://github.com/etsy/statsd) integration settings. + +# If true, metrics_allow_list and metrics_block_list will use regex pattern matching +# anywhere within the metric name instead of only prefix matching at the start of the name. +# +# Variable: AIRFLOW__METRICS__METRICS_USE_PATTERN_MATCH +# +metrics_use_pattern_match = False + +# Configure an allow list (comma separated string) to send only certain metrics. +# If metrics_use_pattern_match is false, match only the exact metric name prefix. +# If metrics_use_pattern_match is true, provide regex patterns to match. +# +# Example: metrics_allow_list = "scheduler,executor,dagrun" or "^scheduler,^executor,heartbeat|timeout" +# +# Variable: AIRFLOW__METRICS__METRICS_ALLOW_LIST +# +metrics_allow_list = + +# Configure a block list (comma separated string) to block certain metrics from being emitted. +# If metrics_allow_list and metrics_block_list are both configured, metrics_block_list is ignored. +# If metrics_use_pattern_match is false, match only the exact metric name prefix. +# If metrics_use_pattern_match is true, provide regex patterns to match. +# +# Example: metrics_block_list = "scheduler,executor,dagrun" or "^scheduler,^executor,heartbeat|timeout" +# +# Variable: AIRFLOW__METRICS__METRICS_BLOCK_LIST +# +metrics_block_list = + +# Enables sending metrics to StatsD. +# +# Variable: AIRFLOW__METRICS__STATSD_ON +# +statsd_on = False + +# Specifies the host address where the StatsD daemon (or server) is running +# +# Variable: AIRFLOW__METRICS__STATSD_HOST +# +statsd_host = localhost + +# Specifies the port on which the StatsD daemon (or server) is listening to +# +# Variable: AIRFLOW__METRICS__STATSD_PORT +# +statsd_port = 8125 + +# Defines the namespace for all metrics sent from Airflow to StatsD +# +# Variable: AIRFLOW__METRICS__STATSD_PREFIX +# +statsd_prefix = airflow + +# A function that validate the StatsD stat name, apply changes to the stat name if necessary and return +# the transformed stat name. +# +# The function should have the following signature: +# def func_name(stat_name: str) -> str: +# +# Variable: AIRFLOW__METRICS__STAT_NAME_HANDLER +# +stat_name_handler = + +# To enable datadog integration to send airflow metrics. +# +# Variable: AIRFLOW__METRICS__STATSD_DATADOG_ENABLED +# +statsd_datadog_enabled = False + +# List of datadog tags attached to all metrics(e.g: key1:value1,key2:value2) +# +# Variable: AIRFLOW__METRICS__STATSD_DATADOG_TAGS +# +statsd_datadog_tags = + +# Set to False to disable metadata tags for some of the emitted metrics +# +# Variable: AIRFLOW__METRICS__STATSD_DATADOG_METRICS_TAGS +# +statsd_datadog_metrics_tags = True + +# If you want to utilise your own custom StatsD client set the relevant +# module path below. +# Note: The module path must exist on your PYTHONPATH for Airflow to pick it up +# +# Variable: AIRFLOW__METRICS__STATSD_CUSTOM_CLIENT_PATH +# +# statsd_custom_client_path = + +# If you want to avoid sending all the available metrics tags to StatsD, +# you can configure a block list of prefixes (comma separated) to filter out metric tags +# that start with the elements of the list (e.g: "job_id,run_id") +# +# Example: statsd_disabled_tags = job_id,run_id,dag_id,task_id +# +# Variable: AIRFLOW__METRICS__STATSD_DISABLED_TAGS +# +statsd_disabled_tags = job_id,run_id + +# To enable sending Airflow metrics with StatsD-Influxdb tagging convention. +# +# Variable: AIRFLOW__METRICS__STATSD_INFLUXDB_ENABLED +# +statsd_influxdb_enabled = False + +# Enables sending metrics to OpenTelemetry. +# +# Variable: AIRFLOW__METRICS__OTEL_ON +# +otel_on = False + +# Specifies the hostname or IP address of the OpenTelemetry Collector to which Airflow sends +# metrics and traces. +# +# Variable: AIRFLOW__METRICS__OTEL_HOST +# +otel_host = localhost + +# Specifies the port of the OpenTelemetry Collector that is listening to. +# +# Variable: AIRFLOW__METRICS__OTEL_PORT +# +otel_port = 8889 + +# The prefix for the Airflow metrics. +# +# Variable: AIRFLOW__METRICS__OTEL_PREFIX +# +otel_prefix = airflow + +# Defines the interval, in milliseconds, at which Airflow sends batches of metrics and traces +# to the configured OpenTelemetry Collector. +# +# Variable: AIRFLOW__METRICS__OTEL_INTERVAL_MILLISECONDS +# +otel_interval_milliseconds = 60000 + +# If True, all metrics are also emitted to the console. Defaults to False. +# +# Variable: AIRFLOW__METRICS__OTEL_DEBUGGING_ON +# +otel_debugging_on = False + +# If True, SSL will be enabled. Defaults to False. +# To establish an HTTPS connection to the OpenTelemetry collector, +# you need to configure the SSL certificate and key within the OpenTelemetry collector's +# config.yml file. +# +# Variable: AIRFLOW__METRICS__OTEL_SSL_ACTIVE +# +otel_ssl_active = False + +[secrets] +# Full class name of secrets backend to enable (will precede env vars and metastore in search path) +# +# Example: backend = airflow.providers.amazon.aws.secrets.systems_manager.SystemsManagerParameterStoreBackend +# +# Variable: AIRFLOW__SECRETS__BACKEND +# +backend = + +# The backend_kwargs param is loaded into a dictionary and passed to __init__ of secrets backend class. +# See documentation for the secrets backend you are using. JSON is expected. +# Example for AWS Systems Manager ParameterStore: +# ``{"connections_prefix": "/airflow/connections", "profile_name": "default"}`` +# +# Variable: AIRFLOW__SECRETS__BACKEND_KWARGS +# +backend_kwargs = + +# .. note:: |experimental| +# +# Enables local caching of Variables, when parsing DAGs only. +# Using this option can make dag parsing faster if Variables are used in top level code, at the expense +# of longer propagation time for changes. +# Please note that this cache concerns only the DAG parsing step. There is no caching in place when DAG +# tasks are run. +# +# Variable: AIRFLOW__SECRETS__USE_CACHE +# +use_cache = False + +# .. note:: |experimental| +# +# When the cache is enabled, this is the duration for which we consider an entry in the cache to be +# valid. Entries are refreshed if they are older than this many seconds. +# It means that when the cache is enabled, this is the maximum amount of time you need to wait to see a +# Variable change take effect. +# +# Variable: AIRFLOW__SECRETS__CACHE_TTL_SECONDS +# +cache_ttl_seconds = 900 + +[cli] +# In what way should the cli access the API. The LocalClient will use the +# database directly, while the json_client will use the api running on the +# webserver +# +# Variable: AIRFLOW__CLI__API_CLIENT +# +api_client = airflow.api.client.local_client + +# If you set web_server_url_prefix, do NOT forget to append it here, ex: +# ``endpoint_url = http://localhost:8080/myroot`` +# So api will look like: ``http://localhost:8080/myroot/api/experimental/...`` +# +# Variable: AIRFLOW__CLI__ENDPOINT_URL +# +endpoint_url = http://localhost:8080 + +[debug] +# Used only with ``DebugExecutor``. If set to ``True`` DAG will fail with first +# failed task. Helpful for debugging purposes. +# +# Variable: AIRFLOW__DEBUG__FAIL_FAST +# +fail_fast = False + +[api] +# Enables the deprecated experimental API. Please note that these APIs do not have access control. +# The authenticated user has full access. +# +# .. warning:: +# +# This `Experimental REST API `__ is +# deprecated since version 2.0. Please consider using +# `the Stable REST API `__. +# For more information on migration, see +# `RELEASE_NOTES.rst `_ +# +# Variable: AIRFLOW__API__ENABLE_EXPERIMENTAL_API +# +enable_experimental_api = False + +# Comma separated list of auth backends to authenticate users of the API. See +# https://airflow.apache.org/docs/apache-airflow/stable/security/api.html for possible values. +# ("airflow.api.auth.backend.default" allows all requests for historic reasons) +# +# Variable: AIRFLOW__API__AUTH_BACKENDS +# +auth_backends = airflow.api.auth.backend.session + +# Used to set the maximum page limit for API requests. If limit passed as param +# is greater than maximum page limit, it will be ignored and maximum page limit value +# will be set as the limit +# +# Variable: AIRFLOW__API__MAXIMUM_PAGE_LIMIT +# +maximum_page_limit = 100 + +# Used to set the default page limit when limit param is zero or not provided in API +# requests. Otherwise if positive integer is passed in the API requests as limit, the +# smallest number of user given limit or maximum page limit is taken as limit. +# +# Variable: AIRFLOW__API__FALLBACK_PAGE_LIMIT +# +fallback_page_limit = 100 + +# The intended audience for JWT token credentials used for authorization. This value must match on the client and server sides. If empty, audience will not be tested. +# +# Example: google_oauth2_audience = project-id-random-value.apps.googleusercontent.com +# +# Variable: AIRFLOW__API__GOOGLE_OAUTH2_AUDIENCE +# +google_oauth2_audience = + +# Path to Google Cloud Service Account key file (JSON). If omitted, authorization based on +# `the Application Default Credentials +# `__ will +# be used. +# +# Example: google_key_path = /files/service-account-json +# +# Variable: AIRFLOW__API__GOOGLE_KEY_PATH +# +google_key_path = + +# Used in response to a preflight request to indicate which HTTP +# headers can be used when making the actual request. This header is +# the server side response to the browser's +# Access-Control-Request-Headers header. +# +# Variable: AIRFLOW__API__ACCESS_CONTROL_ALLOW_HEADERS +# +access_control_allow_headers = + +# Specifies the method or methods allowed when accessing the resource. +# +# Variable: AIRFLOW__API__ACCESS_CONTROL_ALLOW_METHODS +# +access_control_allow_methods = + +# Indicates whether the response can be shared with requesting code from the given origins. +# Separate URLs with space. +# +# Variable: AIRFLOW__API__ACCESS_CONTROL_ALLOW_ORIGINS +# +access_control_allow_origins = + +# Indicates whether the *xcomEntries* endpoint supports the *deserialize* +# flag. If set to False, setting this flag in a request would result in a +# 400 Bad Request error. +# +# Variable: AIRFLOW__API__ENABLE_XCOM_DESERIALIZE_SUPPORT +# +enable_xcom_deserialize_support = False + +[lineage] +# what lineage backend to use +# +# Variable: AIRFLOW__LINEAGE__BACKEND +# +backend = + +[operators] +# The default owner assigned to each new operator, unless +# provided explicitly or passed via ``default_args`` +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_OWNER +# +default_owner = airflow + +# The default value of attribute "deferrable" in operators and sensors. +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE +# +default_deferrable = false + +# Indicates the default number of CPU units allocated to each operator when no specific CPU request +# is specified in the operator's configuration +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_CPUS +# +default_cpus = 1 + +# Indicates the default number of RAM allocated to each operator when no specific RAM request +# is specified in the operator's configuration +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_RAM +# +default_ram = 512 + +# Indicates the default number of disk storage allocated to each operator when no specific disk request +# is specified in the operator's configuration +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_DISK +# +default_disk = 512 + +# Indicates the default number of GPUs allocated to each operator when no specific GPUs request +# is specified in the operator's configuration +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_GPUS +# +default_gpus = 0 + +# Default queue that tasks get assigned to and that worker listen on. +# +# Variable: AIRFLOW__OPERATORS__DEFAULT_QUEUE +# +default_queue = default + +# Is allowed to pass additional/unused arguments (args, kwargs) to the BaseOperator operator. +# If set to False, an exception will be thrown, otherwise only the console message will be displayed. +# +# Variable: AIRFLOW__OPERATORS__ALLOW_ILLEGAL_ARGUMENTS +# +allow_illegal_arguments = False + +[webserver] +# The message displayed when a user attempts to execute actions beyond their authorised privileges. +# +# Variable: AIRFLOW__WEBSERVER__ACCESS_DENIED_MESSAGE +# +access_denied_message = Access is Denied + +# Path of webserver config file used for configuring the webserver parameters +# +# Variable: AIRFLOW__WEBSERVER__CONFIG_FILE +# +config_file = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/webserver_config.py + +# The base url of your website: Airflow cannot guess what domain or CNAME you are using. +# This is used to create links in the Log Url column in the Browse - Task Instances menu, +# as well as in any automated emails sent by Airflow that contain links to your webserver. +# +# Variable: AIRFLOW__WEBSERVER__BASE_URL +# +base_url = http://localhost:8080 + +# Default timezone to display all dates in the UI, can be UTC, system, or +# any IANA timezone string (e.g. Europe/Amsterdam). If left empty the +# default value of core/default_timezone will be used +# +# Example: default_ui_timezone = America/New_York +# +# Variable: AIRFLOW__WEBSERVER__DEFAULT_UI_TIMEZONE +# +default_ui_timezone = UTC + +# The ip specified when starting the web server +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_HOST +# +web_server_host = 0.0.0.0 + +# The port on which to run the web server +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_PORT +# +web_server_port = 8080 + +# Paths to the SSL certificate and key for the web server. When both are +# provided SSL will be enabled. This does not change the web server port. +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_SSL_CERT +# +web_server_ssl_cert = + +# Paths to the SSL certificate and key for the web server. When both are +# provided SSL will be enabled. This does not change the web server port. +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_SSL_KEY +# +web_server_ssl_key = + +# The type of backend used to store web session data, can be `database` or `securecookie`. For the +# `database` backend, sessions are store in the database (in `session` table) and they can be +# managed there (for example when you reset password of the user, all sessions for that user are +# deleted). For the `securecookie` backend, sessions are stored in encrypted cookies on the client +# side. The `securecookie` mechanism is 'lighter' than database backend, but sessions are not deleted +# when you reset password of the user, which means that other than waiting for expiry time, the only +# way to invalidate all sessions for a user is to change secret_key and restart webserver (which +# also invalidates and logs out all other user's sessions). +# +# When you are using `database` backend, make sure to keep your database session table small +# by periodically running `airflow db clean --table session` command, especially if you have +# automated API calls that will create a new session for each call rather than reuse the sessions +# stored in browser cookies. +# +# Example: session_backend = securecookie +# +# Variable: AIRFLOW__WEBSERVER__SESSION_BACKEND +# +session_backend = database + +# Number of seconds the webserver waits before killing gunicorn master that doesn't respond +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_MASTER_TIMEOUT +# +web_server_master_timeout = 120 + +# Number of seconds the gunicorn webserver waits before timing out on a worker +# +# Variable: AIRFLOW__WEBSERVER__WEB_SERVER_WORKER_TIMEOUT +# +web_server_worker_timeout = 120 + +# Number of workers to refresh at a time. When set to 0, worker refresh is +# disabled. When nonzero, airflow periodically refreshes webserver workers by +# bringing up new ones and killing old ones. +# +# Variable: AIRFLOW__WEBSERVER__WORKER_REFRESH_BATCH_SIZE +# +worker_refresh_batch_size = 1 + +# Number of seconds to wait before refreshing a batch of workers. +# +# Variable: AIRFLOW__WEBSERVER__WORKER_REFRESH_INTERVAL +# +worker_refresh_interval = 6000 + +# If set to True, Airflow will track files in plugins_folder directory. When it detects changes, +# then reload the gunicorn. If set to True, gunicorn starts without preloading, which is slower, uses +# more memory, and may cause race conditions. Avoid setting this to True in production. +# +# Variable: AIRFLOW__WEBSERVER__RELOAD_ON_PLUGIN_CHANGE +# +reload_on_plugin_change = False + +# Secret key used to run your flask app. It should be as random as possible. However, when running +# more than 1 instances of webserver, make sure all of them use the same ``secret_key`` otherwise +# one of them will error with "CSRF session token is missing". +# The webserver key is also used to authorize requests to Celery workers when logs are retrieved. +# The token generated using the secret key has a short expiry time though - make sure that time on +# ALL the machines that you run airflow components on is synchronized (for example using ntpd) +# otherwise you might get "forbidden" errors when the logs are accessed. +# +# Variable: AIRFLOW__WEBSERVER__SECRET_KEY +# +secret_key = 0CtJ89VCX7KuAmQmhTycEA== + +# Number of workers to run the Gunicorn web server +# +# Variable: AIRFLOW__WEBSERVER__WORKERS +# +workers = 4 + +# The worker class gunicorn should use. Choices include +# sync (default), eventlet, gevent. Note when using gevent you might also want to set the +# "_AIRFLOW_PATCH_GEVENT" environment variable to "1" to make sure gevent patching is done as +# early as possible. +# +# Variable: AIRFLOW__WEBSERVER__WORKER_CLASS +# +worker_class = sync + +# Log files for the gunicorn webserver. '-' means log to stderr. +# +# Variable: AIRFLOW__WEBSERVER__ACCESS_LOGFILE +# +access_logfile = - + +# Log files for the gunicorn webserver. '-' means log to stderr. +# +# Variable: AIRFLOW__WEBSERVER__ERROR_LOGFILE +# +error_logfile = - + +# Access log format for gunicorn webserver. +# default format is %%(h)s %%(l)s %%(u)s %%(t)s "%%(r)s" %%(s)s %%(b)s "%%(f)s" "%%(a)s" +# documentation - https://docs.gunicorn.org/en/stable/settings.html#access-log-format +# +# Variable: AIRFLOW__WEBSERVER__ACCESS_LOGFORMAT +# +access_logformat = + +# Expose the configuration file in the web server. Set to "non-sensitive-only" to show all values +# except those that have security implications. "True" shows all values. "False" hides the +# configuration completely. +# +# Variable: AIRFLOW__WEBSERVER__EXPOSE_CONFIG +# +expose_config = False + +# Expose hostname in the web server +# +# Variable: AIRFLOW__WEBSERVER__EXPOSE_HOSTNAME +# +expose_hostname = False + +# Expose stacktrace in the web server +# +# Variable: AIRFLOW__WEBSERVER__EXPOSE_STACKTRACE +# +expose_stacktrace = False + +# Default DAG view. Valid values are: ``grid``, ``graph``, ``duration``, ``gantt``, ``landing_times`` +# +# Variable: AIRFLOW__WEBSERVER__DAG_DEFAULT_VIEW +# +dag_default_view = grid + +# Default DAG orientation. Valid values are: +# ``LR`` (Left->Right), ``TB`` (Top->Bottom), ``RL`` (Right->Left), ``BT`` (Bottom->Top) +# +# Variable: AIRFLOW__WEBSERVER__DAG_ORIENTATION +# +dag_orientation = LR + +# Sorting order in grid view. Valid values are: ``topological``, ``hierarchical_alphabetical`` +# +# Variable: AIRFLOW__WEBSERVER__GRID_VIEW_SORTING_ORDER +# +grid_view_sorting_order = topological + +# The amount of time (in secs) webserver will wait for initial handshake +# while fetching logs from other worker machine +# +# Variable: AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC +# +log_fetch_timeout_sec = 5 + +# Time interval (in secs) to wait before next log fetching. +# +# Variable: AIRFLOW__WEBSERVER__LOG_FETCH_DELAY_SEC +# +log_fetch_delay_sec = 2 + +# Distance away from page bottom to enable auto tailing. +# +# Variable: AIRFLOW__WEBSERVER__LOG_AUTO_TAILING_OFFSET +# +log_auto_tailing_offset = 30 + +# Animation speed for auto tailing log display. +# +# Variable: AIRFLOW__WEBSERVER__LOG_ANIMATION_SPEED +# +log_animation_speed = 1000 + +# By default, the webserver shows paused DAGs. Flip this to hide paused +# DAGs by default +# +# Variable: AIRFLOW__WEBSERVER__HIDE_PAUSED_DAGS_BY_DEFAULT +# +hide_paused_dags_by_default = False + +# Consistent page size across all listing views in the UI +# +# Variable: AIRFLOW__WEBSERVER__PAGE_SIZE +# +page_size = 100 + +# Define the color of navigation bar +# +# Variable: AIRFLOW__WEBSERVER__NAVBAR_COLOR +# +navbar_color = #fff + +# Define the color of text in the navigation bar +# +# Variable: AIRFLOW__WEBSERVER__NAVBAR_TEXT_COLOR +# +navbar_text_color = #51504f + +# Define the color of navigation bar links when hovered +# +# Variable: AIRFLOW__WEBSERVER__NAVBAR_HOVER_COLOR +# +navbar_hover_color = #eee + +# Define the color of text in the navigation bar when hovered +# +# Variable: AIRFLOW__WEBSERVER__NAVBAR_TEXT_HOVER_COLOR +# +navbar_text_hover_color = #51504f + +# Define the color of the logo text +# +# Variable: AIRFLOW__WEBSERVER__NAVBAR_LOGO_TEXT_COLOR +# +navbar_logo_text_color = #51504f + +# Default dagrun to show in UI +# +# Variable: AIRFLOW__WEBSERVER__DEFAULT_DAG_RUN_DISPLAY_NUMBER +# +default_dag_run_display_number = 25 + +# Enable werkzeug ``ProxyFix`` middleware for reverse proxy +# +# Variable: AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX +# +enable_proxy_fix = False + +# Number of values to trust for ``X-Forwarded-For``. +# More info: https://werkzeug.palletsprojects.com/en/0.16.x/middleware/proxy_fix/ +# +# Variable: AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR +# +proxy_fix_x_for = 1 + +# Number of values to trust for ``X-Forwarded-Proto`` +# +# Variable: AIRFLOW__WEBSERVER__PROXY_FIX_X_PROTO +# +proxy_fix_x_proto = 1 + +# Number of values to trust for ``X-Forwarded-Host`` +# +# Variable: AIRFLOW__WEBSERVER__PROXY_FIX_X_HOST +# +proxy_fix_x_host = 1 + +# Number of values to trust for ``X-Forwarded-Port`` +# +# Variable: AIRFLOW__WEBSERVER__PROXY_FIX_X_PORT +# +proxy_fix_x_port = 1 + +# Number of values to trust for ``X-Forwarded-Prefix`` +# +# Variable: AIRFLOW__WEBSERVER__PROXY_FIX_X_PREFIX +# +proxy_fix_x_prefix = 1 + +# Set secure flag on session cookie +# +# Variable: AIRFLOW__WEBSERVER__COOKIE_SECURE +# +cookie_secure = False + +# Set samesite policy on session cookie +# +# Variable: AIRFLOW__WEBSERVER__COOKIE_SAMESITE +# +cookie_samesite = Lax + +# Default setting for wrap toggle on DAG code and TI log views. +# +# Variable: AIRFLOW__WEBSERVER__DEFAULT_WRAP +# +default_wrap = False + +# Allow the UI to be rendered in a frame +# +# Variable: AIRFLOW__WEBSERVER__X_FRAME_ENABLED +# +x_frame_enabled = True + +# Send anonymous user activity to your analytics tool +# choose from google_analytics, segment, metarouter, or matomo +# +# Variable: AIRFLOW__WEBSERVER__ANALYTICS_TOOL +# +# analytics_tool = + +# Unique ID of your account in the analytics tool +# +# Variable: AIRFLOW__WEBSERVER__ANALYTICS_ID +# +# analytics_id = + +# Your instances url, only applicable to Matomo. +# +# Example: analytics_url = https://your.matomo.instance.com/ +# +# Variable: AIRFLOW__WEBSERVER__ANALYTICS_URL +# +# analytics_url = + +# 'Recent Tasks' stats will show for old DagRuns if set +# +# Variable: AIRFLOW__WEBSERVER__SHOW_RECENT_STATS_FOR_COMPLETED_RUNS +# +show_recent_stats_for_completed_runs = True + +# Update FAB permissions and sync security manager roles +# on webserver startup +# +# Variable: AIRFLOW__WEBSERVER__UPDATE_FAB_PERMS +# +update_fab_perms = True + +# The UI cookie lifetime in minutes. User will be logged out from UI after +# ``session_lifetime_minutes`` of non-activity +# +# Variable: AIRFLOW__WEBSERVER__SESSION_LIFETIME_MINUTES +# +session_lifetime_minutes = 43200 + +# Sets a custom page title for the DAGs overview page and site title for all pages +# +# Variable: AIRFLOW__WEBSERVER__INSTANCE_NAME +# +# instance_name = + +# Whether the custom page title for the DAGs overview page contains any Markup language +# +# Variable: AIRFLOW__WEBSERVER__INSTANCE_NAME_HAS_MARKUP +# +instance_name_has_markup = False + +# How frequently, in seconds, the DAG data will auto-refresh in graph or grid view +# when auto-refresh is turned on +# +# Variable: AIRFLOW__WEBSERVER__AUTO_REFRESH_INTERVAL +# +auto_refresh_interval = 3 + +# Boolean for displaying warning for publicly viewable deployment +# +# Variable: AIRFLOW__WEBSERVER__WARN_DEPLOYMENT_EXPOSURE +# +warn_deployment_exposure = True + +# Comma separated string of view events to exclude from dag audit view. +# All other events will be added minus the ones passed here. +# The audit logs in the db will not be affected by this parameter. +# +# Example: audit_view_excluded_events = cli_task_run,running,success +# +# Variable: AIRFLOW__WEBSERVER__AUDIT_VIEW_EXCLUDED_EVENTS +# +# audit_view_excluded_events = + +# Comma separated string of view events to include in dag audit view. +# If passed, only these events will populate the dag audit view. +# The audit logs in the db will not be affected by this parameter. +# +# Example: audit_view_included_events = dagrun_cleared,failed +# +# Variable: AIRFLOW__WEBSERVER__AUDIT_VIEW_INCLUDED_EVENTS +# +# audit_view_included_events = + +# Boolean for running SwaggerUI in the webserver. +# +# Variable: AIRFLOW__WEBSERVER__ENABLE_SWAGGER_UI +# +enable_swagger_ui = True + +# Boolean for running Internal API in the webserver. +# +# Variable: AIRFLOW__WEBSERVER__RUN_INTERNAL_API +# +run_internal_api = False + +# Boolean for enabling rate limiting on authentication endpoints. +# +# Variable: AIRFLOW__WEBSERVER__AUTH_RATE_LIMITED +# +auth_rate_limited = True + +# Rate limit for authentication endpoints. +# +# Variable: AIRFLOW__WEBSERVER__AUTH_RATE_LIMIT +# +auth_rate_limit = 5 per 40 second + +# The caching algorithm used by the webserver. Must be a valid hashlib function name. +# +# Example: caching_hash_method = sha256 +# +# Variable: AIRFLOW__WEBSERVER__CACHING_HASH_METHOD +# +caching_hash_method = md5 + +# Behavior of the trigger DAG run button for DAGs without params. False to skip and trigger +# without displaying a form to add a dag_run.conf, True to always display the form. +# The form is displayed always if parameters are defined. +# +# Variable: AIRFLOW__WEBSERVER__SHOW_TRIGGER_FORM_IF_NO_PARAMS +# +show_trigger_form_if_no_params = False + +# Number of recent DAG run configurations in the selector on the trigger web form. +# +# Example: num_recent_configurations_for_trigger = 10 +# +# Variable: AIRFLOW__WEBSERVER__NUM_RECENT_CONFIGURATIONS_FOR_TRIGGER +# +num_recent_configurations_for_trigger = 5 + +# A DAG author is able to provide any raw HTML into ``doc_md`` or params description in +# ``description_md`` for text formatting. This is including potentially unsafe javascript. +# Displaying the DAG or trigger form in web UI provides the DAG author the potential to +# inject malicious code into clients browsers. To ensure the web UI is safe by default, +# raw HTML is disabled by default. If you trust your DAG authors, you can enable HTML +# support in markdown by setting this option to True. +# +# This parameter also enables the deprecated fields ``description_html`` and +# ``custom_html_form`` in DAG params until the feature is removed in a future version. +# +# Example: allow_raw_html_descriptions = False +# +# Variable: AIRFLOW__WEBSERVER__ALLOW_RAW_HTML_DESCRIPTIONS +# +allow_raw_html_descriptions = False + +# The maximum size of the request payload (in MB) that can be sent. +# +# Variable: AIRFLOW__WEBSERVER__ALLOWED_PAYLOAD_SIZE +# +allowed_payload_size = 1.0 + +# Require confirmation when changing a DAG in the web UI. This is to prevent accidental changes +# to a DAG that may be running on sensitive environments like production. +# When set to True, confirmation dialog will be shown when a user tries to Pause/Unpause, Trigger a DAG +# +# Variable: AIRFLOW__WEBSERVER__REQUIRE_CONFIRMATION_DAG_CHANGE +# +require_confirmation_dag_change = False + +[email] +# Configuration email backend and whether to +# send email alerts on retry or failure + +# Email backend to use +# +# Variable: AIRFLOW__EMAIL__EMAIL_BACKEND +# +email_backend = airflow.utils.email.send_email_smtp + +# Email connection to use +# +# Variable: AIRFLOW__EMAIL__EMAIL_CONN_ID +# +email_conn_id = smtp_default + +# Whether email alerts should be sent when a task is retried +# +# Variable: AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY +# +default_email_on_retry = True + +# Whether email alerts should be sent when a task failed +# +# Variable: AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE +# +default_email_on_failure = True + +# File that will be used as the template for Email subject (which will be rendered using Jinja2). +# If not set, Airflow uses a base template. +# +# Example: subject_template = /path/to/my_subject_template_file +# +# Variable: AIRFLOW__EMAIL__SUBJECT_TEMPLATE +# +# subject_template = + +# File that will be used as the template for Email content (which will be rendered using Jinja2). +# If not set, Airflow uses a base template. +# +# Example: html_content_template = /path/to/my_html_content_template_file +# +# Variable: AIRFLOW__EMAIL__HTML_CONTENT_TEMPLATE +# +# html_content_template = + +# Email address that will be used as sender address. +# It can either be raw email or the complete address in a format ``Sender Name `` +# +# Example: from_email = Airflow +# +# Variable: AIRFLOW__EMAIL__FROM_EMAIL +# +# from_email = + +# ssl context to use when using SMTP and IMAP SSL connections. By default, the context is "default" +# which sets it to ``ssl.create_default_context()`` which provides the right balance between +# compatibility and security, it however requires that certificates in your operating system are +# updated and that SMTP/IMAP servers of yours have valid certificates that have corresponding public +# keys installed on your machines. You can switch it to "none" if you want to disable checking +# of the certificates, but it is not recommended as it allows MITM (man-in-the-middle) attacks +# if your infrastructure is not sufficiently secured. It should only be set temporarily while you +# are fixing your certificate configuration. This can be typically done by upgrading to newer +# version of the operating system you run Airflow components on,by upgrading/refreshing proper +# certificates in the OS or by updating certificates for your mail servers. +# +# Example: ssl_context = default +# +# Variable: AIRFLOW__EMAIL__SSL_CONTEXT +# +ssl_context = default + +[smtp] +# If you want airflow to send emails on retries, failure, and you want to use +# the airflow.utils.email.send_email_smtp function, you have to configure an +# smtp server here + +# Specifies the host server address used by Airflow when sending out email notifications via SMTP. +# +# Variable: AIRFLOW__SMTP__SMTP_HOST +# +smtp_host = localhost + +# Determines whether to use the STARTTLS command when connecting to the SMTP server. +# +# Variable: AIRFLOW__SMTP__SMTP_STARTTLS +# +smtp_starttls = True + +# Determines whether to use an SSL connection when talking to the SMTP server. +# +# Variable: AIRFLOW__SMTP__SMTP_SSL +# +smtp_ssl = False + +# Username to authenticate when connecting to smtp server. +# +# Example: smtp_user = airflow +# +# Variable: AIRFLOW__SMTP__SMTP_USER +# +# smtp_user = + +# Password to authenticate when connecting to smtp server. +# +# Example: smtp_password = airflow +# +# Variable: AIRFLOW__SMTP__SMTP_PASSWORD +# +# smtp_password = + +# Defines the port number on which Airflow connects to the SMTP server to send email notifications. +# +# Variable: AIRFLOW__SMTP__SMTP_PORT +# +smtp_port = 25 + +# Specifies the default "from" email address used when Airflow sends email notifications. +# +# Variable: AIRFLOW__SMTP__SMTP_MAIL_FROM +# +smtp_mail_from = airflow@example.com + +# Determines the maximum time (in seconds) the Apache Airflow system will wait for a +# connection to the SMTP server to be established. +# +# Variable: AIRFLOW__SMTP__SMTP_TIMEOUT +# +smtp_timeout = 30 + +# Defines the maximum number of times Airflow will attempt to connect to the SMTP server. +# +# Variable: AIRFLOW__SMTP__SMTP_RETRY_LIMIT +# +smtp_retry_limit = 5 + +[sentry] +# Sentry (https://docs.sentry.io) integration. Here you can supply +# additional configuration options based on the Python platform. See: +# https://docs.sentry.io/error-reporting/configuration/?platform=python. +# Unsupported options: ``integrations``, ``in_app_include``, ``in_app_exclude``, +# ``ignore_errors``, ``before_breadcrumb``, ``transport``. + +# Enable error reporting to Sentry +# +# Variable: AIRFLOW__SENTRY__SENTRY_ON +# +sentry_on = false + +# +# Variable: AIRFLOW__SENTRY__SENTRY_DSN +# +sentry_dsn = + +# Dotted path to a before_send function that the sentry SDK should be configured to use. +# +# Variable: AIRFLOW__SENTRY__BEFORE_SEND +# +# before_send = + +[scheduler] +# Task instances listen for external kill signal (when you clear tasks +# from the CLI or the UI), this defines the frequency at which they should +# listen (in seconds). +# +# Variable: AIRFLOW__SCHEDULER__JOB_HEARTBEAT_SEC +# +job_heartbeat_sec = 5 + +# The scheduler constantly tries to trigger new tasks (look at the +# scheduler section in the docs for more information). This defines +# how often the scheduler should run (in seconds). +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_HEARTBEAT_SEC +# +scheduler_heartbeat_sec = 5 + +# The frequency (in seconds) at which the LocalTaskJob should send heartbeat signals to the +# scheduler to notify it's still alive. If this value is set to 0, the heartbeat interval will default +# to the value of scheduler_zombie_task_threshold. +# +# Variable: AIRFLOW__SCHEDULER__LOCAL_TASK_JOB_HEARTBEAT_SEC +# +local_task_job_heartbeat_sec = 0 + +# The number of times to try to schedule each DAG file +# -1 indicates unlimited number +# +# Variable: AIRFLOW__SCHEDULER__NUM_RUNS +# +num_runs = -1 + +# Controls how long the scheduler will sleep between loops, but if there was nothing to do +# in the loop. i.e. if it scheduled something then it will start the next loop +# iteration straight away. +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_IDLE_SLEEP_TIME +# +scheduler_idle_sleep_time = 1 + +# Number of seconds after which a DAG file is parsed. The DAG file is parsed every +# ``min_file_process_interval`` number of seconds. Updates to DAGs are reflected after +# this interval. Keeping this number low will increase CPU usage. +# +# Variable: AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL +# +min_file_process_interval = 30 + +# How often (in seconds) to check for stale DAGs (DAGs which are no longer present in +# the expected files) which should be deactivated, as well as datasets that are no longer +# referenced and should be marked as orphaned. +# +# Variable: AIRFLOW__SCHEDULER__PARSING_CLEANUP_INTERVAL +# +parsing_cleanup_interval = 60 + +# How long (in seconds) to wait after we have re-parsed a DAG file before deactivating stale +# DAGs (DAGs which are no longer present in the expected files). The reason why we need +# this threshold is to account for the time between when the file is parsed and when the +# DAG is loaded. The absolute maximum that this could take is `dag_file_processor_timeout`, +# but when you have a long timeout configured, it results in a significant delay in the +# deactivation of stale dags. +# +# Variable: AIRFLOW__SCHEDULER__STALE_DAG_THRESHOLD +# +stale_dag_threshold = 50 + +# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes. +# +# Variable: AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL +# +dag_dir_list_interval = 300 + +# How often should stats be printed to the logs. Setting to 0 will disable printing stats +# +# Variable: AIRFLOW__SCHEDULER__PRINT_STATS_INTERVAL +# +print_stats_interval = 30 + +# How often (in seconds) should pool usage stats be sent to StatsD (if statsd_on is enabled) +# +# Variable: AIRFLOW__SCHEDULER__POOL_METRICS_INTERVAL +# +pool_metrics_interval = 5.0 + +# If the last scheduler heartbeat happened more than scheduler_health_check_threshold +# ago (in seconds), scheduler is considered unhealthy. +# This is used by the health check in the "/health" endpoint and in `airflow jobs check` CLI +# for SchedulerJob. +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_HEALTH_CHECK_THRESHOLD +# +scheduler_health_check_threshold = 30 + +# When you start a scheduler, airflow starts a tiny web server +# subprocess to serve a health check if this is set to True +# +# Variable: AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK +# +enable_health_check = False + +# When you start a scheduler, airflow starts a tiny web server +# subprocess to serve a health check on this host +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_HEALTH_CHECK_SERVER_HOST +# +scheduler_health_check_server_host = 0.0.0.0 + +# When you start a scheduler, airflow starts a tiny web server +# subprocess to serve a health check on this port +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_HEALTH_CHECK_SERVER_PORT +# +scheduler_health_check_server_port = 8974 + +# How often (in seconds) should the scheduler check for orphaned tasks and SchedulerJobs +# +# Variable: AIRFLOW__SCHEDULER__ORPHANED_TASKS_CHECK_INTERVAL +# +orphaned_tasks_check_interval = 300.0 + +# Determines the directory where logs for the child processes of the scheduler will be stored +# +# Variable: AIRFLOW__SCHEDULER__CHILD_PROCESS_LOG_DIRECTORY +# +child_process_log_directory = /Users/burakkurt/data_engineering_projects/e-commerce-analytics/airflow/logs/scheduler + +# Local task jobs periodically heartbeat to the DB. If the job has +# not heartbeat in this many seconds, the scheduler will mark the +# associated task instance as failed and will re-schedule the task. +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULER_ZOMBIE_TASK_THRESHOLD +# +scheduler_zombie_task_threshold = 300 + +# How often (in seconds) should the scheduler check for zombie tasks. +# +# Variable: AIRFLOW__SCHEDULER__ZOMBIE_DETECTION_INTERVAL +# +zombie_detection_interval = 10.0 + +# Turn off scheduler catchup by setting this to ``False``. +# Default behavior is unchanged and +# Command Line Backfills still work, but the scheduler +# will not do scheduler catchup if this is ``False``, +# however it can be set on a per DAG basis in the +# DAG definition (catchup) +# +# Variable: AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT +# +catchup_by_default = True + +# Setting this to True will make first task instance of a task +# ignore depends_on_past setting. A task instance will be considered +# as the first task instance of a task when there is no task instance +# in the DB with an execution_date earlier than it., i.e. no manual marking +# success will be needed for a newly added task to be scheduled. +# +# Variable: AIRFLOW__SCHEDULER__IGNORE_FIRST_DEPENDS_ON_PAST_BY_DEFAULT +# +ignore_first_depends_on_past_by_default = True + +# This changes the batch size of queries in the scheduling main loop. +# This should not be greater than ``core.parallelism``. +# If this is too high, SQL query performance may be impacted by +# complexity of query predicate, and/or excessive locking. +# Additionally, you may hit the maximum allowable query length for your db. +# Set this to 0 to use the value of ``core.parallelism`` +# +# Variable: AIRFLOW__SCHEDULER__MAX_TIS_PER_QUERY +# +max_tis_per_query = 16 + +# Should the scheduler issue ``SELECT ... FOR UPDATE`` in relevant queries. +# If this is set to False then you should not run more than a single +# scheduler at once +# +# Variable: AIRFLOW__SCHEDULER__USE_ROW_LEVEL_LOCKING +# +use_row_level_locking = True + +# Max number of DAGs to create DagRuns for per scheduler loop. +# +# Variable: AIRFLOW__SCHEDULER__MAX_DAGRUNS_TO_CREATE_PER_LOOP +# +max_dagruns_to_create_per_loop = 10 + +# How many DagRuns should a scheduler examine (and lock) when scheduling +# and queuing tasks. +# +# Variable: AIRFLOW__SCHEDULER__MAX_DAGRUNS_PER_LOOP_TO_SCHEDULE +# +max_dagruns_per_loop_to_schedule = 20 + +# Should the Task supervisor process perform a "mini scheduler" to attempt to schedule more tasks of the +# same DAG. Leaving this on will mean tasks in the same DAG execute quicker, but might starve out other +# dags in some circumstances +# +# Variable: AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION +# +schedule_after_task_execution = True + +# The scheduler reads dag files to extract the airflow modules that are going to be used, +# and imports them ahead of time to avoid having to re-do it for each parsing process. +# This flag can be set to False to disable this behavior in case an airflow module needs to be freshly +# imported each time (at the cost of increased DAG parsing time). +# +# Variable: AIRFLOW__SCHEDULER__PARSING_PRE_IMPORT_MODULES +# +parsing_pre_import_modules = True + +# The scheduler can run multiple processes in parallel to parse dags. +# This defines how many processes will run. +# +# Variable: AIRFLOW__SCHEDULER__PARSING_PROCESSES +# +parsing_processes = 2 + +# One of ``modified_time``, ``random_seeded_by_host`` and ``alphabetical``. +# The scheduler will list and sort the dag files to decide the parsing order. +# +# * ``modified_time``: Sort by modified time of the files. This is useful on large scale to parse the +# recently modified DAGs first. +# * ``random_seeded_by_host``: Sort randomly across multiple Schedulers but with same order on the +# same host. This is useful when running with Scheduler in HA mode where each scheduler can +# parse different DAG files. +# * ``alphabetical``: Sort by filename +# +# Variable: AIRFLOW__SCHEDULER__FILE_PARSING_SORT_MODE +# +file_parsing_sort_mode = modified_time + +# Whether the dag processor is running as a standalone process or it is a subprocess of a scheduler +# job. +# +# Variable: AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR +# +standalone_dag_processor = False + +# Only applicable if `[scheduler]standalone_dag_processor` is true and callbacks are stored +# in database. Contains maximum number of callbacks that are fetched during a single loop. +# +# Variable: AIRFLOW__SCHEDULER__MAX_CALLBACKS_PER_LOOP +# +max_callbacks_per_loop = 20 + +# Only applicable if `[scheduler]standalone_dag_processor` is true. +# Time in seconds after which dags, which were not updated by Dag Processor are deactivated. +# +# Variable: AIRFLOW__SCHEDULER__DAG_STALE_NOT_SEEN_DURATION +# +dag_stale_not_seen_duration = 600 + +# Turn off scheduler use of cron intervals by setting this to False. +# DAGs submitted manually in the web UI or with trigger_dag will still run. +# +# Variable: AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE +# +use_job_schedule = True + +# Allow externally triggered DagRuns for Execution Dates in the future +# Only has effect if schedule_interval is set to None in DAG +# +# Variable: AIRFLOW__SCHEDULER__ALLOW_TRIGGER_IN_FUTURE +# +allow_trigger_in_future = False + +# How often to check for expired trigger requests that have not run yet. +# +# Variable: AIRFLOW__SCHEDULER__TRIGGER_TIMEOUT_CHECK_INTERVAL +# +trigger_timeout_check_interval = 15 + +# Amount of time a task can be in the queued state before being retried or set to failed. +# +# Variable: AIRFLOW__SCHEDULER__TASK_QUEUED_TIMEOUT +# +task_queued_timeout = 600.0 + +# How often to check for tasks that have been in the queued state for +# longer than `[scheduler] task_queued_timeout`. +# +# Variable: AIRFLOW__SCHEDULER__TASK_QUEUED_TIMEOUT_CHECK_INTERVAL +# +task_queued_timeout_check_interval = 120.0 + +# The run_id pattern used to verify the validity of user input to the run_id parameter when +# triggering a DAG. This pattern cannot change the pattern used by scheduler to generate run_id +# for scheduled DAG runs or DAG runs triggered without changing the run_id parameter. +# +# Variable: AIRFLOW__SCHEDULER__ALLOWED_RUN_ID_PATTERN +# +allowed_run_id_pattern = ^[A-Za-z0-9_.~:+-]+$ + +# Whether to create DAG runs that span an interval or one single point in time for cron schedules, when +# a cron string is provided to `schedule` argument of a DAG. If True, +# CronDataIntervalTimetable is used, which is the legacy Airflow behavior suitable +# for DAGs with well-defined data_interval you get contiguous intervals from the end of the previous +# interval up to the scheduled datetime. If False, CronTriggerTimetable is used, +# which is closer to the behavior of cron itself. +# +# Notably, for CronTriggerTimetable, the logical_date is the same as the time the DAG Run will try to +# schedule, while for CronDataIntervalTimetable, the logical_date is the beginning of the data interval, +# but the DAG Run will try to schedule at the end of the data interval. For more differences +# between the two Timetables, see +# https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/timetable.html#differences-between-the-two-cron-timetables +# +# Variable: AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVALS +# +create_cron_data_intervals = True + +[triggerer] +# How many triggers a single Triggerer will run at once, by default. +# +# Variable: AIRFLOW__TRIGGERER__DEFAULT_CAPACITY +# +default_capacity = 1000 + +# How often to heartbeat the Triggerer job to ensure it hasn't been killed. +# +# Variable: AIRFLOW__TRIGGERER__JOB_HEARTBEAT_SEC +# +job_heartbeat_sec = 5 + +# If the last triggerer heartbeat happened more than triggerer_health_check_threshold +# ago (in seconds), triggerer is considered unhealthy. +# This is used by the health check in the "/health" endpoint and in `airflow jobs check` CLI +# for TriggererJob. +# +# Variable: AIRFLOW__TRIGGERER__TRIGGERER_HEALTH_CHECK_THRESHOLD +# +triggerer_health_check_threshold = 30 + +[kerberos] +# Location of your ccache file once kinit has been performed. +# +# Variable: AIRFLOW__KERBEROS__CCACHE +# +ccache = /tmp/airflow_krb5_ccache + +# gets augmented with fqdn +# +# Variable: AIRFLOW__KERBEROS__PRINCIPAL +# +principal = airflow + +# Determines the frequency at which initialization or re-initialization processes occur. +# +# Variable: AIRFLOW__KERBEROS__REINIT_FREQUENCY +# +reinit_frequency = 3600 + +# Path to the kinit executable +# +# Variable: AIRFLOW__KERBEROS__KINIT_PATH +# +kinit_path = kinit + +# Designates the path to the Kerberos keytab file for the Airflow user +# +# Variable: AIRFLOW__KERBEROS__KEYTAB +# +keytab = airflow.keytab + +# Allow to disable ticket forwardability. +# +# Variable: AIRFLOW__KERBEROS__FORWARDABLE +# +forwardable = True + +# Allow to remove source IP from token, useful when using token behind NATted Docker host. +# +# Variable: AIRFLOW__KERBEROS__INCLUDE_IP +# +include_ip = True + +[sensors] +# Sensor default timeout, 7 days by default (7 * 24 * 60 * 60). +# +# Variable: AIRFLOW__SENSORS__DEFAULT_TIMEOUT +# +default_timeout = 604800 + +[common.io] +# Common IO configuration section + +# Path to a location on object storage where XComs can be stored in url format. +# +# Example: xcom_objectstorage_path = s3://conn_id@bucket/path +# +# Variable: AIRFLOW__COMMON.IO__XCOM_OBJECTSTORAGE_PATH +# +xcom_objectstorage_path = + +# Threshold in bytes for storing XComs in object storage. -1 means always store in the +# database. 0 means always store in object storage. Any positive number means +# it will be stored in object storage if the size of the value is greater than the threshold. +# +# Example: xcom_objectstorage_threshold = 1000000 +# +# Variable: AIRFLOW__COMMON.IO__XCOM_OBJECTSTORAGE_THRESHOLD +# +xcom_objectstorage_threshold = -1 + +# Compression algorithm to use when storing XComs in object storage. Supported algorithms +# are a.o.: snappy, zip, gzip, bz2, and lzma. If not specified, no compression will be used. +# Note that the compression algorithm must be available in the Python installation (e.g. +# python-snappy for snappy). Zip, gz, bz2 are available by default. +# +# Example: xcom_objectstorage_compression = gz +# +# Variable: AIRFLOW__COMMON.IO__XCOM_OBJECTSTORAGE_COMPRESSION +# +xcom_objectstorage_compression = + +[fab] +# This section contains configs specific to FAB provider. + +# Boolean for enabling rate limiting on authentication endpoints. +# +# Variable: AIRFLOW__FAB__AUTH_RATE_LIMITED +# +auth_rate_limited = True + +# Rate limit for authentication endpoints. +# +# Variable: AIRFLOW__FAB__AUTH_RATE_LIMIT +# +auth_rate_limit = 5 per 40 second + +# Update FAB permissions and sync security manager roles +# on webserver startup +# +# Variable: AIRFLOW__FAB__UPDATE_FAB_PERMS +# +update_fab_perms = True + +[imap] +# Options for IMAP provider. + +# ssl_context = + +[smtp_provider] +# Options for SMTP provider. + +# ssl context to use when using SMTP and IMAP SSL connections. By default, the context is "default" +# which sets it to ``ssl.create_default_context()`` which provides the right balance between +# compatibility and security, it however requires that certificates in your operating system are +# updated and that SMTP/IMAP servers of yours have valid certificates that have corresponding public +# keys installed on your machines. You can switch it to "none" if you want to disable checking +# of the certificates, but it is not recommended as it allows MITM (man-in-the-middle) attacks +# if your infrastructure is not sufficiently secured. It should only be set temporarily while you +# are fixing your certificate configuration. This can be typically done by upgrading to newer +# version of the operating system you run Airflow components on,by upgrading/refreshing proper +# certificates in the OS or by updating certificates for your mail servers. +# +# If you do not set this option explicitly, it will use Airflow "email.ssl_context" configuration, +# but if this configuration is not present, it will use "default" value. +# +# Example: ssl_context = default +# +# Variable: AIRFLOW__SMTP_PROVIDER__SSL_CONTEXT +# +# ssl_context = + +# Allows overriding of the standard templated email subject line when the SmtpNotifier is used. +# Must provide a path to the template. +# +# Example: templated_email_subject_path = path/to/override/email_subject.html +# +# Variable: AIRFLOW__SMTP_PROVIDER__TEMPLATED_EMAIL_SUBJECT_PATH +# +# templated_email_subject_path = + +# Allows overriding of the standard templated email path when the SmtpNotifier is used. Must provide +# a path to the template. +# +# Example: templated_html_content_path = path/to/override/email.html +# +# Variable: AIRFLOW__SMTP_PROVIDER__TEMPLATED_HTML_CONTENT_PATH +# +# templated_html_content_path = + diff --git a/airflow/dags/ecom_pipeline.py b/airflow/dags/ecom_pipeline.py new file mode 100644 index 00000000..f205fea7 --- /dev/null +++ b/airflow/dags/ecom_pipeline.py @@ -0,0 +1,97 @@ +from datetime import datetime, timedelta +import subprocess +import os +from airflow import DAG +from airflow.operators.python import PythonOperator, BranchPythonOperator +from airflow.operators.bash import BashOperator +from airflow.operators.empty import EmptyOperator + +PROJECT_ROOT = "/Users/burakkurt/data_engineering_projects/e-commerce-analytics" +DBT_VENV = PROJECT_ROOT + "/.venv/bin" +DBT_DIR = PROJECT_ROOT + "/ecom" +RAW_DIR = PROJECT_ROOT + "/data/raw" + +default_args = { + "owner": "burak", + "retries": 1, + "retry_delay": timedelta(minutes=2), + "email_on_failure": False, +} + +def check_freshness(**context): + import time + orders_path = RAW_DIR + "/orders.csv" + if not os.path.exists(orders_path): + print("STALE: orders.csv not found") + return "notify_stale" + age_hours = (time.time() - os.path.getmtime(orders_path)) / 3600 + print("orders.csv age: %.1f hours" % age_hours) + if age_hours > 24: + print("STALE: skipping build") + return "notify_stale" + print("FRESH: proceeding with build") + return "load_raw_to_bigquery" + +def load_raw(**context): + result = subprocess.run( + [PROJECT_ROOT + "/.venv/bin/python", + PROJECT_ROOT + "/scripts/load_bigquery.py"], + capture_output=True, text=True + ) + print(result.stdout) + if result.returncode != 0: + raise Exception("load_bigquery.py failed: " + result.stderr) + +def run_dbt(**context): + result = subprocess.run( + [DBT_VENV + "/dbt", "build", + "--profiles-dir", ".", + "--target", "bigquery"], + capture_output=True, text=True, + cwd=DBT_DIR + ) + print(result.stdout[-3000:]) + if result.returncode != 0: + raise Exception("dbt build failed: " + result.stderr) + +with DAG( + dag_id="ecom_daily_pipeline", + default_args=default_args, + description="Daily ecom pipeline: freshness check, BQ load, dbt build", + schedule_interval="0 6 * * *", + start_date=datetime(2024, 10, 1), + catchup=False, + tags=["ecom", "dbt", "bigquery"], +) as dag: + + start = EmptyOperator(task_id="start") + + freshness_check = BranchPythonOperator( + task_id="check_source_freshness", + python_callable=check_freshness, + ) + + load_raw_task = PythonOperator( + task_id="load_raw_to_bigquery", + python_callable=load_raw, + ) + + dbt_build = PythonOperator( + task_id="dbt_build_bigquery", + python_callable=run_dbt, + ) + + notify_success = BashOperator( + task_id="notify_success", + bash_command='echo "Pipeline complete"', + trigger_rule="none_failed_min_one_success", + ) + + notify_stale = BashOperator( + task_id="notify_stale", + bash_command='echo "SKIPPED: source data stale"', + ) + + start >> freshness_check + freshness_check >> load_raw_task >> dbt_build >> notify_success + freshness_check >> notify_stale diff --git a/airflow/standalone_admin_password.txt b/airflow/standalone_admin_password.txt new file mode 100644 index 00000000..17b39a5a --- /dev/null +++ b/airflow/standalone_admin_password.txt @@ -0,0 +1 @@ +8UNzpahcqDWAFhHx \ No newline at end of file diff --git a/airflow/webserver_config.py b/airflow/webserver_config.py new file mode 100644 index 00000000..3048bb21 --- /dev/null +++ b/airflow/webserver_config.py @@ -0,0 +1,132 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Default configuration for the Airflow webserver.""" + +from __future__ import annotations + +import os + +from flask_appbuilder.const import AUTH_DB + +# from airflow.www.fab_security.manager import AUTH_LDAP +# from airflow.www.fab_security.manager import AUTH_OAUTH +# from airflow.www.fab_security.manager import AUTH_OID +# from airflow.www.fab_security.manager import AUTH_REMOTE_USER + + +basedir = os.path.abspath(os.path.dirname(__file__)) + +# Flask-WTF flag for CSRF +WTF_CSRF_ENABLED = True +WTF_CSRF_TIME_LIMIT = None + +# ---------------------------------------------------- +# AUTHENTICATION CONFIG +# ---------------------------------------------------- +# For details on how to set up each of the following authentication, see +# http://flask-appbuilder.readthedocs.io/en/latest/security.html# authentication-methods +# for details. + +# The authentication type +# AUTH_OID : Is for OpenID +# AUTH_DB : Is for database +# AUTH_LDAP : Is for LDAP +# AUTH_REMOTE_USER : Is for using REMOTE_USER from web server +# AUTH_OAUTH : Is for OAuth +AUTH_TYPE = AUTH_DB + +# Uncomment to setup Full admin role name +# AUTH_ROLE_ADMIN = 'Admin' + +# Uncomment and set to desired role to enable access without authentication +# AUTH_ROLE_PUBLIC = 'Viewer' + +# Will allow user self registration +# AUTH_USER_REGISTRATION = True + +# The recaptcha it's automatically enabled for user self registration is active and the keys are necessary +# RECAPTCHA_PRIVATE_KEY = PRIVATE_KEY +# RECAPTCHA_PUBLIC_KEY = PUBLIC_KEY + +# Config for Flask-Mail necessary for user self registration +# MAIL_SERVER = 'smtp.gmail.com' +# MAIL_USE_TLS = True +# MAIL_USERNAME = 'yourappemail@gmail.com' +# MAIL_PASSWORD = 'passwordformail' +# MAIL_DEFAULT_SENDER = 'sender@gmail.com' + +# The default user self registration role +# AUTH_USER_REGISTRATION_ROLE = "Public" + +# When using OAuth Auth, uncomment to setup provider(s) info +# Google OAuth example: +# OAUTH_PROVIDERS = [{ +# 'name':'google', +# 'token_key':'access_token', +# 'icon':'fa-google', +# 'remote_app': { +# 'api_base_url':'https://www.googleapis.com/oauth2/v2/', +# 'client_kwargs':{ +# 'scope': 'email profile' +# }, +# 'access_token_url':'https://accounts.google.com/o/oauth2/token', +# 'authorize_url':'https://accounts.google.com/o/oauth2/auth', +# 'request_token_url': None, +# 'client_id': GOOGLE_KEY, +# 'client_secret': GOOGLE_SECRET_KEY, +# } +# }] + +# When using LDAP Auth, setup the ldap server +# AUTH_LDAP_SERVER = "ldap://ldapserver.new" + +# When using OpenID Auth, uncomment to setup OpenID providers. +# example for OpenID authentication +# OPENID_PROVIDERS = [ +# { 'name': 'Yahoo', 'url': 'https://me.yahoo.com' }, +# { 'name': 'AOL', 'url': 'http://openid.aol.com/' }, +# { 'name': 'Flickr', 'url': 'http://www.flickr.com/' }, +# { 'name': 'MyOpenID', 'url': 'https://www.myopenid.com' }] + +# ---------------------------------------------------- +# Theme CONFIG +# ---------------------------------------------------- +# Flask App Builder comes up with a number of predefined themes +# that you can use for Apache Airflow. +# http://flask-appbuilder.readthedocs.io/en/latest/customizing.html#changing-themes +# Please make sure to remove "navbar_color" configuration from airflow.cfg +# in order to fully utilize the theme. (or use that property in conjunction with theme) +# APP_THEME = "bootstrap-theme.css" # default bootstrap +# APP_THEME = "amelia.css" +# APP_THEME = "cerulean.css" +# APP_THEME = "cosmo.css" +# APP_THEME = "cyborg.css" +# APP_THEME = "darkly.css" +# APP_THEME = "flatly.css" +# APP_THEME = "journal.css" +# APP_THEME = "lumen.css" +# APP_THEME = "paper.css" +# APP_THEME = "readable.css" +# APP_THEME = "sandstone.css" +# APP_THEME = "simplex.css" +# APP_THEME = "slate.css" +# APP_THEME = "solar.css" +# APP_THEME = "spacelab.css" +# APP_THEME = "superhero.css" +# APP_THEME = "united.css" +# APP_THEME = "yeti.css" diff --git a/ecom/dbt_project.yml b/ecom/dbt_project.yml index b0a753cc..5271b7d4 100644 --- a/ecom/dbt_project.yml +++ b/ecom/dbt_project.yml @@ -1,16 +1,20 @@ -name: ecom +name: 'ecom' version: '1.0.0' config-version: 2 -profile: ecom +profile: 'ecom' model-paths: ["models"] +seed-paths: ["seeds"] test-paths: ["tests"] +analysis-paths: ["analyses"] macro-paths: ["macros"] snapshot-paths: ["snapshots"] target-path: "target" -clean-targets: ["target", "dbt_packages"] +clean-targets: + - "target" + - "dbt_packages" models: ecom: diff --git a/ecom/models/marts/_schema.yml b/ecom/models/marts/_schema.yml index 2067c45f..9fabe783 100644 --- a/ecom/models/marts/_schema.yml +++ b/ecom/models/marts/_schema.yml @@ -84,7 +84,7 @@ models: - name: dim_customer_scd2 description: > SCD Type 2 customer dimension. One row per customer per time period. - Use valid_from/valid_to to join to fact_orders at the correct point in time. + Use effective_date/expiry_date to join to fact_orders at the correct point in time. is_current = TRUE for the version active today. columns: - name: customer_scd_key @@ -94,9 +94,41 @@ models: - name: customer_id data_tests: - not_null - - name: valid_from + - name: effective_date data_tests: - not_null - - name: version_number + - name: is_current_flag data_tests: - not_null + + - name: dim_campaign + description: "Campaign dimension — flash sales, coupons, seasonal, loyalty." + columns: + - name: campaign_key + description: "Surrogate key. -1 used in fact_orders when no campaign applies." + tests: [not_null, unique] + - name: campaign_id + tests: [not_null, unique] + - name: campaign_type + description: "Type: flash_sale, coupon, seasonal, loyalty." + + - name: dim_seller_scd2 + description: "Type 2 seller dimension. Tracks tier and rating changes using effective_date/expiry_date/is_current_flag — Trendyol pattern." + columns: + - name: seller_scd_key + description: "Surrogate key — unique per seller version." + tests: [not_null, unique] + - name: seller_id + tests: [not_null] + - name: is_current_flag + description: "True for the currently active version." + + - name: fact_seller_daily + description: "Pre-aggregated daily seller performance — one row per seller per day. Incremental." + columns: + - name: seller_key + tests: [not_null] + - name: stat_date + tests: [not_null] + - name: gmv + description: "Total line revenue for this seller on this day." diff --git a/ecom/models/marts/dim_campaign.sql b/ecom/models/marts/dim_campaign.sql new file mode 100644 index 00000000..39d26d7d --- /dev/null +++ b/ecom/models/marts/dim_campaign.sql @@ -0,0 +1,15 @@ +with final as ( + select + row_number() over (order by campaign_id) as campaign_key, + campaign_id, + campaign_name, + campaign_type, + start_date, + end_date, + discount_pct, + is_active, + end_date - start_date as duration_days + from {{ ref('stg_campaigns') }} +) + +select * from final diff --git a/ecom/models/marts/dim_customer_scd2.sql b/ecom/models/marts/dim_customer_scd2.sql index bf01c614..d747b011 100644 --- a/ecom/models/marts/dim_customer_scd2.sql +++ b/ecom/models/marts/dim_customer_scd2.sql @@ -1,149 +1,31 @@ -/* - dim_customer_scd2 — Slowly Changing Dimension Type 2 - ------------------------------------------------------- - - WHY SCD TYPE 2? - --------------- - A customer's city and tier change over time. If we just overwrite the - current value (Type 1), we lose history. That means a report asking - "what was this customer's tier when they placed this order?" becomes - unanswerable — we only know what they are *now*. - - Type 2 solves this by keeping every historical version of a customer - as a separate row, each stamped with valid_from and valid_to dates. - - GRAIN: one row per customer per time period (version). - - KEY COLUMNS: - customer_scd_key surrogate key — unique per VERSION (not per customer) - customer_id natural key — same across all versions of one customer - valid_from date this version became active - valid_to date this version was superseded (NULL = still active) - is_current TRUE for the version active right now - version_number 1 = original record, 2 = first change, etc. - - HOW IT WORKS (step by step): - 1. Take the original signup record for every customer. - 2. Union with all change events from stg_customer_changes. - 3. Use LEAD() window function to look ahead to the next event date - for the same customer — that becomes the current row's valid_to. - 4. The last version per customer has no next event, so valid_to = NULL - and is_current = TRUE. - - FACT TABLE USAGE: - To get the customer dimension that was active when an order was placed: - JOIN dim_customer_scd2 scd - ON fact.customer_id = scd.customer_id - AND fact.order_date >= scd.valid_from - AND (fact.order_date < scd.valid_to OR scd.valid_to IS NULL) -*/ - -with customers as ( - select - customer_id, - city, - customer_tier, - signup_date as effective_date, - first_name, - last_name, - email - from {{ ref('stg_customers') }} -), - -changes as ( +with changes as ( select customer_id, city, customer_tier, effective_date, - null as first_name, - null as last_name, - null as email - from {{ ref('stg_customer_changes') }} -), - --- combine original records and change events into one timeline -all_versions as ( - select * from customers - union all - select * from changes -), - --- carry forward name/email from the original record --- (change events don't re-supply these) -with_context as ( - select - customer_id, - effective_date, - -- coalesce fills nulls in change rows with the original values - coalesce( - city, - first_value(city) over ( - partition by customer_id - order by effective_date - rows between unbounded preceding and current row - ) - ) as city, - coalesce( - customer_tier, - first_value(customer_tier) over ( - partition by customer_id - order by effective_date - rows between unbounded preceding and current row - ) - ) as customer_tier, - first_value(first_name) over ( - partition by customer_id order by effective_date - ) as first_name, - first_value(last_name) over ( - partition by customer_id order by effective_date - ) as last_name, - first_value(email) over ( - partition by customer_id order by effective_date - ) as email, - row_number() over ( - partition by customer_id - order by effective_date - ) as version_number - from all_versions -), - --- apply LEAD() to compute valid_to from the next version's start date -with_validity as ( - select - customer_id, - first_name, - last_name, - email, - city, - customer_tier, - effective_date as valid_from, lead(effective_date) over ( partition by customer_id order by effective_date - ) as valid_to, -- NULL on the current version - version_number - from with_context + ) as expiry_date + from {{ ref('stg_customer_changes') }} ), final as ( select - -- surrogate key: unique per customer VERSION row_number() over ( - order by customer_id, valid_from - ) as customer_scd_key, + order by customer_id, effective_date + ) as customer_scd_key, customer_id, - first_name, - last_name, - first_name || ' ' || last_name as full_name, - email, city, customer_tier, - valid_from, - valid_to, - valid_to is null as is_current, - version_number - from with_validity + effective_date, + expiry_date, + case + when expiry_date is null then true + else false + end as is_current_flag + from changes ) select * from final diff --git a/ecom/models/marts/dim_date.sql b/ecom/models/marts/dim_date.sql index 2d6b636b..3f2da23a 100644 --- a/ecom/models/marts/dim_date.sql +++ b/ecom/models/marts/dim_date.sql @@ -1,22 +1,40 @@ -/* - dim_date - -------- - A generated date dimension covering the full data range (Oct 2024 – Mar 2025). - No source table needed — built purely from DuckDB's generate_series(). +{% if target.type == 'bigquery' %} - date_key is an integer in YYYYMMDD format (e.g. 20241015). - This is the foreign key pattern used in fact_orders. +with date_spine as ( + select date + from unnest( + generate_date_array('2024-10-01', '2025-03-31', interval 1 day) + ) as date +), - Why a date dimension? - - Lets analysts filter/group by year, month, quarter, weekday without - writing date functions in every query. - - A single tested source of date logic = single source of truth. -*/ +final as ( + select + cast(format_date('%Y%m%d', date) as int64) as date_key, + date, + extract(year from date) as year, + extract(month from date) as month_number, + format_date('%B', date) as month_name, + extract(quarter from date) as quarter, + concat('Q', cast(extract(quarter from date) as string)) as quarter_label, + extract(dayofweek from date) as day_of_week, + format_date('%A', date) as day_name, + extract(dayofweek from date) in (1, 7) as is_weekend, + extract(week from date) as week_of_year + from date_spine +) + +select * from final + +{% else %} with date_spine as ( select unnest( - generate_series(date '2024-10-01', date '2025-03-31', interval '1 day') + generate_series( + date '2024-10-01', + date '2025-03-31', + interval '1 day' + ) )::date as date ), @@ -37,3 +55,5 @@ final as ( ) select * from final + +{% endif %} diff --git a/ecom/models/marts/dim_seller_scd2.sql b/ecom/models/marts/dim_seller_scd2.sql new file mode 100644 index 00000000..39eb9e16 --- /dev/null +++ b/ecom/models/marts/dim_seller_scd2.sql @@ -0,0 +1,31 @@ +with changes as ( + select + seller_id, + seller_tier, + rating, + change_date as effective_date, + lead(change_date) over ( + partition by seller_id + order by change_date + ) as expiry_date + from {{ ref('stg_seller_changes') }} +), + +final as ( + select + row_number() over ( + order by seller_id, effective_date + ) as seller_scd_key, + seller_id, + seller_tier, + rating, + effective_date, + expiry_date, + case + when expiry_date is null then true + else false + end as is_current_flag + from changes +) + +select * from final diff --git a/ecom/models/marts/fact_orders.sql b/ecom/models/marts/fact_orders.sql index de1cfe34..31c796de 100644 --- a/ecom/models/marts/fact_orders.sql +++ b/ecom/models/marts/fact_orders.sql @@ -2,24 +2,16 @@ fact_orders ----------- Grain: one row per order LINE ITEM (most granular fact table). - - Why line-item grain and not order grain? - - An order can have multiple products — line-item grain lets you analyse - revenue, margin, and flash-sale rates per product, category, and seller. - - You can always roll up to order level with COUNT(DISTINCT order_id). - - Measures (the numbers): - quantity, unit_price, list_price, line_revenue, discount_pct - - Foreign keys (the context — join to dims for labels): - date_key → dim_date - customer_key → dim_customer - product_key → dim_product - seller_key → dim_seller - - Degenerate dimensions (IDs with no dim table): - order_id, order_item_id — kept directly in the fact table + Materialization: incremental (unique_key = order_item_id) + New in Phase 1 upgrade: + - campaign_key FK to dim_campaign (-1 sentinel when no campaign) + - order_date column added for incremental watermark + - converted from table to incremental */ +{{ config( + materialized='incremental', + unique_key='order_item_id' +) }} with orders as ( select * from {{ ref('stg_orders') }} @@ -41,34 +33,36 @@ dim_seller as ( select seller_key, seller_id from {{ ref('dim_seller') }} ), +dim_campaign as ( + select campaign_key, campaign_id from {{ ref('dim_campaign') }} +), + joined as ( select - -- degenerate dimensions (natural keys kept in fact) oi.order_item_id, o.order_id, - - -- foreign keys to dimensions + {% if target.type == 'bigquery' %} + cast(format_date('%Y%m%d', cast(o.order_date as date)) as int64) as date_key, + {% else %} cast(strftime(o.order_date::date, '%Y%m%d') as integer) as date_key, + {% endif %} dc.customer_key, dp.product_key, ds.seller_key, - - -- measures + coalesce(dcam.campaign_key, -1) as campaign_key, oi.quantity, oi.unit_price, oi.list_price, oi.discount_pct, oi.line_revenue, oi.is_flash_sale, - - -- order-level attributes (non-additive context) o.order_status, o.payment_method, o.shipping_city, o.platform, o.discount_amount as order_discount_amount, - o.coupon_code - + o.coupon_code, + o.order_date from order_items oi inner join orders o on oi.order_id = o.order_id @@ -78,6 +72,39 @@ joined as ( on oi.product_id = dp.product_id left join dim_seller ds on oi.seller_id = ds.seller_id + left join dim_campaign dcam + on o.campaign_id = dcam.campaign_id +) + +{% if is_incremental() %} +, +watermark as ( + select coalesce( + max( + {% if target.type == 'bigquery' %} + cast(order_date as date) + {% else %} + order_date::date + {% endif %} + ), + {% if target.type == 'bigquery' %} + date '1900-01-01' + {% else %} + '1900-01-01'::date + {% endif %} + ) as max_date + from {{ this }} ) +{% endif %} -select * from joined +select joined.* +from joined +{% if is_incremental() %} +cross join watermark +where + {% if target.type == 'bigquery' %} + cast(joined.order_date as date) > watermark.max_date + {% else %} + joined.order_date::date > watermark.max_date + {% endif %} +{% endif %} diff --git a/ecom/models/marts/fact_seller_daily.sql b/ecom/models/marts/fact_seller_daily.sql new file mode 100644 index 00000000..aaf241d2 --- /dev/null +++ b/ecom/models/marts/fact_seller_daily.sql @@ -0,0 +1,43 @@ +{{ config( + materialized='incremental', + unique_key=['seller_id', 'stat_date'] +) }} + +with source as ( + select * from {{ ref('stg_seller_daily_stats') }} +), + +dim_seller as ( + select seller_key, seller_id + from {{ ref('dim_seller') }} +), + +dim_date as ( + select date_key, date + from {{ ref('dim_date') }} +), + +joined as ( + select + ds.seller_key, + s.seller_id, + dd.date_key, + s.stat_date, + s.gmv, + s.total_orders, + s.cancelled_orders, + s.returned_orders, + s.total_items, + s.cancellation_rate, + s.return_rate, + s.avg_rating + from source s + join dim_seller ds on s.seller_id = ds.seller_id + join dim_date dd on s.stat_date = dd.date +) + +select * from joined + +{% if is_incremental() %} + where stat_date > (select max(stat_date) from {{ this }}) +{% endif %} diff --git a/ecom/models/staging/_schema.yml b/ecom/models/staging/_schema.yml index 071a3478..0c8b5a1d 100644 --- a/ecom/models/staging/_schema.yml +++ b/ecom/models/staging/_schema.yml @@ -122,3 +122,29 @@ models: - name: effective_date data_tests: - not_null + + - name: stg_campaigns + description: "Campaign records — flash sales, coupons, seasonal, loyalty." + columns: + - name: campaign_id + tests: [not_null, unique] + - name: campaign_type + tests: + - accepted_values: + values: ['flash_sale', 'coupon', 'seasonal', 'loyalty'] + + - name: stg_seller_changes + description: "Change event log for seller SCD2. Source for dim_seller_scd2." + columns: + - name: seller_id + tests: [not_null] + - name: change_date + tests: [not_null] + + - name: stg_seller_daily_stats + description: "Pre-aggregated daily seller metrics. Source for fact_seller_daily." + columns: + - name: seller_id + tests: [not_null] + - name: stat_date + tests: [not_null] diff --git a/ecom/models/staging/_sources.yml b/ecom/models/staging/_sources.yml index bfd2ed91..f22dd938 100644 --- a/ecom/models/staging/_sources.yml +++ b/ecom/models/staging/_sources.yml @@ -19,3 +19,6 @@ sources: description: "Raw app/web sessions — 20% anonymous (null customer_id)" - name: customer_changes description: "Customer attribute change events — city moves and tier upgrades" + - name: campaigns + - name: seller_changes + - name: seller_daily_stats diff --git a/ecom/models/staging/stg_campaigns.sql b/ecom/models/staging/stg_campaigns.sql new file mode 100644 index 00000000..102bea16 --- /dev/null +++ b/ecom/models/staging/stg_campaigns.sql @@ -0,0 +1,17 @@ +with source as ( + select * from {{ source('raw', 'campaigns') }} +), + +final as ( + select + campaign_id, + trim(campaign_name) as campaign_name, + lower(campaign_type) as campaign_type, + cast(start_date as date) as start_date, + cast(end_date as date) as end_date, + cast(discount_pct as numeric) as discount_pct, + cast(is_active as boolean) as is_active + from source +) + +select * from final diff --git a/ecom/models/staging/stg_order_items.sql b/ecom/models/staging/stg_order_items.sql index 44ef2d5c..72d7895a 100644 --- a/ecom/models/staging/stg_order_items.sql +++ b/ecom/models/staging/stg_order_items.sql @@ -22,9 +22,9 @@ casted as ( product_id, seller_id, cast(quantity as integer) as quantity, - cast(unit_price as decimal(10,2)) as unit_price, - cast(list_price as decimal(10,2)) as list_price, - cast(discount_pct as decimal(6,4)) as discount_pct + cast(unit_price as numeric) as unit_price, + cast(list_price as numeric) as list_price, + cast(discount_pct as numeric) as discount_pct from source where order_id in (select order_id from valid_orders) ), diff --git a/ecom/models/staging/stg_orders.sql b/ecom/models/staging/stg_orders.sql index 09e6a70d..ea5f60f0 100644 --- a/ecom/models/staging/stg_orders.sql +++ b/ecom/models/staging/stg_orders.sql @@ -1,39 +1,42 @@ /* stg_orders ---------- - - Key challenge: order_date arrives in two formats (ISO and dd/mm/yyyy). - We COALESCE two TRY_STRPTIME calls to handle both safely. - - Filters rows where the date cannot be parsed (data quality guard) - - Filters orphaned orders (customer_id not in stg_customers) - - Casts and standardises order_status and payment_method + Handles two date formats and filters orphaned customer_ids. + Uses target-conditional date parsing for DuckDB and BigQuery compatibility. */ - with source as ( select * from {{ source('raw', 'orders') }} ), +valid_customers as ( + select customer_id from {{ ref('stg_customers') }} +), + date_parsed as ( select order_id, customer_id, - -- Try ISO first, then dd/mm/yyyy — whichever parses wins + {% if target.type == 'bigquery' %} + coalesce( + safe.parse_date('%Y-%m-%d', order_date), + safe.parse_date('%d/%m/%Y', order_date) + ) as order_date, + {% else %} coalesce( try_strptime(order_date, '%Y-%m-%d'), try_strptime(order_date, '%d/%m/%Y') ) as order_date, + {% endif %} order_status, payment_method, shipping_city, - cast(discount_amount as decimal(10,2)) as discount_amount, + cast(discount_amount as numeric) as discount_amount, nullif(trim(coupon_code), '') as coupon_code, - lower(platform) as platform + lower(platform) as platform, + nullif(trim(campaign_id), '') as campaign_id from source ), -valid_customers as ( - select customer_id from {{ ref('stg_customers') }} -), - final as ( select o.order_id, @@ -44,12 +47,11 @@ final as ( o.shipping_city, o.discount_amount, o.coupon_code, - o.platform + o.platform, + o.campaign_id from date_parsed o - -- Drop unparseable dates where o.order_date is not null - -- Drop orphaned customer_ids (referential integrity fix) - and o.customer_id in (select customer_id from valid_customers) + and o.customer_id in (select customer_id from valid_customers) ) select * from final diff --git a/ecom/models/staging/stg_products.sql b/ecom/models/staging/stg_products.sql index dc8a21c7..2777554c 100644 --- a/ecom/models/staging/stg_products.sql +++ b/ecom/models/staging/stg_products.sql @@ -17,11 +17,11 @@ final as ( trim(category) as category, trim(subcategory) as subcategory, seller_id, - cast(list_price as decimal(10,2)) as list_price, - cast(cost_price as decimal(10,2)) as cost_price, + cast(list_price as numeric) as list_price, + cast(cost_price as numeric) as cost_price, round( - (cast(list_price as decimal(10,2)) - cast(cost_price as decimal(10,2))) - / nullif(cast(list_price as decimal(10,2)), 0), + (cast(list_price as numeric) - cast(cost_price as numeric)) + / nullif(cast(list_price as numeric), 0), 4 ) as gross_margin_pct, cast(is_active as boolean) as is_active, diff --git a/ecom/models/staging/stg_seller_changes.sql b/ecom/models/staging/stg_seller_changes.sql new file mode 100644 index 00000000..b456ee49 --- /dev/null +++ b/ecom/models/staging/stg_seller_changes.sql @@ -0,0 +1,14 @@ +with source as ( + select * from {{ source('raw', 'seller_changes') }} +), + +final as ( + select + seller_id, + lower(seller_tier) as seller_tier, + cast(rating as numeric) as rating, + cast(change_date as date) as change_date + from source +) + +select * from final diff --git a/ecom/models/staging/stg_seller_daily_stats.sql b/ecom/models/staging/stg_seller_daily_stats.sql new file mode 100644 index 00000000..bc6926a4 --- /dev/null +++ b/ecom/models/staging/stg_seller_daily_stats.sql @@ -0,0 +1,20 @@ +with source as ( + select * from {{ source('raw', 'seller_daily_stats') }} +), + +final as ( + select + seller_id, + cast(stat_date as date) as stat_date, + cast(gmv as numeric) as gmv, + cast(total_orders as integer) as total_orders, + cast(cancelled_orders as integer) as cancelled_orders, + cast(returned_orders as integer) as returned_orders, + cast(total_items as integer) as total_items, + cast(cancellation_rate as numeric) as cancellation_rate, + cast(return_rate as numeric) as return_rate, + cast(avg_rating as numeric) as avg_rating + from source +) + +select * from final diff --git a/ecom/models/staging/stg_sellers.sql b/ecom/models/staging/stg_sellers.sql index c6720eee..25d0603e 100644 --- a/ecom/models/staging/stg_sellers.sql +++ b/ecom/models/staging/stg_sellers.sql @@ -18,7 +18,7 @@ final as ( city, cast(is_verified as boolean) as is_verified, cast(joined_date as date) as joined_date, - cast(rating as decimal(3,1)) as rating, + cast(rating as numeric) as rating, cast(total_reviews as integer) as total_reviews from source ) diff --git a/ecom/profiles.yml b/ecom/profiles.yml index 802da62b..83cf8c07 100644 --- a/ecom/profiles.yml +++ b/ecom/profiles.yml @@ -5,3 +5,15 @@ ecom: type: duckdb path: "{{ env_var('DBT_DUCKDB_PATH', '../data/ecom.duckdb') }}" threads: 4 + + bigquery: + type: bigquery + method: service-account + project: pc-api-5223742488799605973-592 + dataset: marts + keyfile: /Users/burakkurt/.dbt/dbt-ecom-key.json + location: EU + threads: 4 + timeout_seconds: 300 + priority: interactive + retries: 1 diff --git a/generator/generate.py b/generator/generate.py index bb0d30c0..7c104ff9 100644 --- a/generator/generate.py +++ b/generator/generate.py @@ -266,4 +266,129 @@ def write_csv(rows, filename): print("\nDone. Files written to data/raw/") -print(f"Date range: {iso(START_DATE)} → {iso(END_DATE)}") \ No newline at end of file +print(f"Date range: {iso(START_DATE)} → {iso(END_DATE)}") + +# ── NEW: CAMPAIGNS ──────────────────────────────────────────────────────────── +def generate_campaigns(n=20, start_date=datetime(2024,10,1), end_date=datetime(2025,3,31)): + import random + CAMPAIGN_TYPES = ["flash_sale", "coupon", "seasonal", "loyalty"] + rows = [] + for i in range(1, n+1): + cs = start_date + timedelta(days=random.randint(0, (end_date-start_date).days - 10)) + ce = cs + timedelta(days=random.randint(1, 7)) + if ce > end_date: + ce = end_date + rows.append({ + "campaign_id": f"CAM{i:03d}", + "campaign_name": f"Campaign {i}", + "campaign_type": random.choice(CAMPAIGN_TYPES), + "start_date": cs.strftime("%Y-%m-%d"), + "end_date": ce.strftime("%Y-%m-%d"), + "discount_pct": round(random.uniform(0.05, 0.50), 2), + "is_active": "true", + }) + return rows + + +# ── NEW: SELLER CHANGES ─────────────────────────────────────────────────────── +def generate_seller_changes(seller_ids, n_sellers=60, + start_date=datetime(2024,10,1), + end_date=datetime(2025,3,31)): + SELLER_TIERS = ["standard", "premium", "elite"] + chosen = random.sample(seller_ids, min(n_sellers, len(seller_ids))) + rows = [] + for sid in chosen: + n_events = random.randint(1, 2) + used = set() + for _ in range(n_events): + d = start_date + timedelta(days=random.randint(0, (end_date-start_date).days)) + while d in used: + d = start_date + timedelta(days=random.randint(0, (end_date-start_date).days)) + used.add(d) + rows.append({ + "seller_id": sid, + "seller_tier": random.choice(SELLER_TIERS), + "rating": round(random.uniform(3.0, 5.0), 2), + "change_date": d.strftime("%Y-%m-%d"), + }) + rows.sort(key=lambda r: (r["seller_id"], r["change_date"])) + return rows + + +# ── NEW: SELLER DAILY STATS ─────────────────────────────────────────────────── +def generate_seller_daily_stats(order_items_rows, orders_rows, sellers_rows): + from collections import defaultdict + order_map = {o["order_id"]: o for o in orders_rows} + daily = defaultdict(lambda: { + "gmv": 0.0, "total_orders": set(), + "cancelled": set(), "returned": set(), "total_items": 0 + }) + for oi in order_items_rows: + order = order_map.get(oi["order_id"]) + if not order: + continue + try: + # parse either date format + od = oi.get("order_date") or order.get("order_date", "") + try: + d = datetime.strptime(od, "%Y-%m-%d").strftime("%Y-%m-%d") + except: + d = datetime.strptime(od, "%d/%m/%Y").strftime("%Y-%m-%d") + except: + continue + key = (oi["seller_id"], d) + qty = int(oi.get("quantity", 1)) + up = float(oi.get("unit_price", 0)) + dp = float(oi.get("discount_pct", 0) or 0) + rev = round(up * qty * (1 - dp), 2) + daily[key]["gmv"] += rev + daily[key]["total_orders"].add(oi["order_id"]) + daily[key]["total_items"] += 1 + status = order.get("order_status", "") + if status == "cancelled": + daily[key]["cancelled"].add(oi["order_id"]) + if status == "returned": + daily[key]["returned"].add(oi["order_id"]) + + rows = [] + for (seller_id, stat_date), v in daily.items(): + total = len(v["total_orders"]) + rows.append({ + "seller_id": seller_id, + "stat_date": stat_date, + "gmv": round(v["gmv"], 2), + "total_orders": total, + "cancelled_orders": len(v["cancelled"]), + "returned_orders": len(v["returned"]), + "total_items": v["total_items"], + "cancellation_rate": round(len(v["cancelled"]) / total, 4) if total else 0, + "return_rate": round(len(v["returned"]) / total, 4) if total else 0, + "avg_rating": round(random.uniform(3.0, 5.0), 2), + }) + rows.sort(key=lambda r: (r["seller_id"], r["stat_date"])) + return rows + + +# ── NEW: WRITE HELPER ───────────────────────────────────────────────────────── +def write_csv_extra(filename, rows): + if not rows: + print(f"SKIP {filename} — no rows") + return + out_dir = os.path.join(os.path.dirname(__file__), "..", "data", "raw") + path = os.path.join(out_dir, filename) + with open(path, "w", newline="", encoding="utf-8") as f: + w = csv.DictWriter(f, fieldnames=rows[0].keys()) + w.writeheader() + w.writerows(rows) + print(f"OK {filename:<35} {len(rows):>5} rows") + +# ── CALL NEW GENERATORS ─────────────────────────────────────────────────────── +campaigns = generate_campaigns(20) +write_csv_extra("campaigns.csv", campaigns) + +seller_ids = [s["seller_id"] for s in sellers] +seller_changes = generate_seller_changes(seller_ids, n_sellers=60) +write_csv_extra("seller_changes.csv", seller_changes) + +seller_daily = generate_seller_daily_stats(order_items, orders, sellers) +write_csv_extra("seller_daily_stats.csv", seller_daily) diff --git a/scripts/add_campaign_to_orders.py b/scripts/add_campaign_to_orders.py new file mode 100644 index 00000000..66ba1a1c --- /dev/null +++ b/scripts/add_campaign_to_orders.py @@ -0,0 +1,50 @@ +""" +One-time patch: adds campaign_id column to orders.csv. +After this patch, campaign_id is generated directly in generate.py. +""" +import csv, random, os +from datetime import datetime + +random.seed(42) +RAW_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "raw") + +# read campaigns +camps = [] +with open(os.path.join(RAW_DIR, "campaigns.csv")) as f: + camps = list(csv.DictReader(f)) + +camp_windows = [ + (c["campaign_id"], c["start_date"], c["end_date"]) for c in camps +] + +# read and patch orders +orders_path = os.path.join(RAW_DIR, "orders.csv") +with open(orders_path) as f: + orders = list(csv.DictReader(f)) + +for o in orders: + try: + try: + od = datetime.strptime(o["order_date"], "%Y-%m-%d").date() + except: + od = datetime.strptime(o["order_date"], "%d/%m/%Y").date() + except: + o["campaign_id"] = "" + continue + assigned = "" + for cid, cs, ce in camp_windows: + cs_d = datetime.strptime(cs, "%Y-%m-%d").date() + ce_d = datetime.strptime(ce, "%Y-%m-%d").date() + if cs_d <= od <= ce_d and random.random() < 0.30: + assigned = cid + break + o["campaign_id"] = assigned + +with open(orders_path, "w", newline="") as f: + w = csv.DictWriter(f, fieldnames=orders[0].keys()) + w.writeheader() + w.writerows(orders) + +print(f"Patched {len(orders)} orders with campaign_id") +assigned_count = sum(1 for o in orders if o["campaign_id"]) +print(f" {assigned_count} orders assigned to a campaign") diff --git a/scripts/load_bigquery.py b/scripts/load_bigquery.py new file mode 100644 index 00000000..c7283f07 --- /dev/null +++ b/scripts/load_bigquery.py @@ -0,0 +1,41 @@ +""" +Loads all CSVs from data/raw/ into BigQuery under the 'raw' dataset. +Run this before dbt build --target bigquery. +""" +from google.cloud import bigquery +import os + +PROJECT = "pc-api-5223742488799605973-592" +DATASET = "raw" +RAW_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "raw") + +TABLES = [ + "customers", "sellers", "products", "orders", "order_items", + "sessions", "customer_changes", "campaigns", + "seller_changes", "seller_daily_stats", +] + +client = bigquery.Client(project=PROJECT) + +for table in TABLES: + path = os.path.join(RAW_DIR, f"{table}.csv") + if not os.path.exists(path): + print(f"SKIP {table} — file not found") + continue + + table_id = f"{PROJECT}.{DATASET}.{table}" + job_config = bigquery.LoadJobConfig( + source_format=bigquery.SourceFormat.CSV, + skip_leading_rows=1, + autodetect=True, + write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE, + ) + + with open(path, "rb") as f: + job = client.load_table_from_file(f, table_id, job_config=job_config) + job.result() + + tbl = client.get_table(table_id) + print(f"OK {DATASET}.{table:<25} {tbl.num_rows:>6,} rows") + +print("\nAll tables loaded into BigQuery raw dataset.") diff --git a/scripts/load_raw.py b/scripts/load_raw.py index 577b776c..a11cc7ed 100644 --- a/scripts/load_raw.py +++ b/scripts/load_raw.py @@ -20,6 +20,9 @@ "order_items", "sessions", "customer_changes", + "campaigns", + "seller_changes", + "seller_daily_stats", ] def main():