mkdir airflow_project && cd airflow_project
~/airflow-local$ curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.9.3/docker-compose.yaml'Last updated
mkdir -p ./dags ./logs ./plugins ./configecho -e "AIRFLOW_UID=$(id -u)" > .env# Basic Airflow cluster configuration for CeleryExecutor with Redis and PostgreSQL.
#
# WARNING: This configuration is for local development. Do not use it in a production deployment.
#
# This configuration supports basic configuration using environment variables or an .env file
# The following variables are supported:
#
# AIRFLOW_IMAGE_NAME - Docker image name used to run Airflow.
# Default: apache/airflow:2.9.3
# AIRFLOW_UID - User ID in Airflow containers
# Default: 50000
# AIRFLOW_PROJ_DIR - Base path to which all the files will be volumed.
# Default: .
# Those configurations are useful mostly in case of standalone testing/running Airflow in test/try-out mode
#
# _AIRFLOW_WWW_USER_USERNAME - Username for the administrator account (if requested).
# Default: airflow
# _AIRFLOW_WWW_USER_PASSWORD - Password for the administrator account (if requested).
# Default: airflow
# _PIP_ADDITIONAL_REQUIREMENTS - Additional PIP requirements to add when starting all containers.
# Use this option ONLY for quick checks. Installing requirements at container
# startup is done EVERY TIME the service is started.
# A better way is to build a custom image or extend the official image
# as described in https://airflow.apache.org/docs/docker-stack/build.html.
# Default: ''
#
# Feel free to modify this file to suit your needs.ports: - "9090:8080"AIRFLOW__CORE__LOAD_EXAMPLES: 'false'x-airflow-common:
AIRFLOW__WEBSERVER__BASE_URL: "http://127.0.0.1:9090"docker compose up airflow-initdocker compose run airflow-worker airflow dags listdocker compose run airflow-worker airflow tasks list <dag_id>docker compose run airflow-worker airflow dags test "print-context"docker compose restart airflow-schedulerdocker compose run airflow-worker airflow dags trigger <dag_id>import pendulum
import urllib.request
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
dag_args = dict(
dag_id="stocksense",
start_date=pendulum.now().subtract(days=1),
template_searchpath="/tmp/",
)
def _get_data(output_path, **context):
execution_date = context["logical_date"]
adjusted_date = execution_date.subtract(hours=2)
year = adjusted_date.year
month = adjusted_date.month
day = adjusted_date.day
hour = adjusted_date.hour
url = (
"https://dumps.wikimedia.org/other/pageviews/"
f"{year}/{year}-{month:0>2}/pageviews-{year}{month:0>2}{day:0>2}-{hour:0>2}0000.gz"
)
print(f"url: {url}")
urllib.request.urlretrieve(url, output_path)
print("Data fetched successfully")
def _fetch_pageviews(pagenames, execution_date, **_):
result = dict.fromkeys(pagenames, 0)
with open("/tmp/wikipageviews", "r") as f:
for line in f:
domain_code, page_title, view_counts, _ = line.split(" ")
if domain_code == "en" and page_title in pagenames:
result[page_title] = view_counts
with open("/tmp/postgres_query.sql", "w") as f:
for pagename, pageviewcount in result.items():
f.write(
"INSERT INTO page_view_counts VALUES "
f"('{pagename}', {pageviewcount}, '{execution_date}')"
)
print(result)
with DAG(**dag_args) as dag:
get_data = PythonOperator(
task_id="get_data",
python_callable=_get_data,
op_kwargs={
"output_path": "/tmp/wikipageviews.gz",
}
)
extract_gz = BashOperator(
task_id="extract_gz",
bash_command="gunzip --force /tmp/wikipageviews.gz",
)
fetch_pageviews = PythonOperator(
task_id="fetch_pageviews",
python_callable=_fetch_pageviews,
op_kwargs={
"pagenames": [
"Google",
"Amazon",
"Apple",
"Microsoft",
"Facebook",
],
}
)
write_to_postgres = PostgresOperator(
task_id="write_to_postgres",
postgres_conn_id="my_postgres",
sql="/tmp/postgres_query.sql",
)
get_data >> extract_gz >> fetch_pageviews >> write_to_postgres