diff --git a/README.md b/README.md index d4b03e0..dfe9161 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,6 @@ A collection of delicious docker recipes. - [ ] libreswan - [ ] mitmproxy - [ ] nagios -- [ ] nfs -- [ ] openldap - [ ] openswan - [ ] postfix - [ ] pritunl @@ -45,6 +43,7 @@ A collection of delicious docker recipes. ## Big Data +- [x] airflow - [x] kafka-arm - [x] kafka-manager - [x] presto @@ -264,7 +263,6 @@ A collection of delicious docker recipes. - [x] confluentinc/cp-kafka-rest - [x] streamsets/datacollector - [x] cachethq/docker -- [x] puckel/docker-airflow - [x] drone/drone - [x] drupal - [x] elastalert diff --git a/airflow/Dockerfile b/airflow/Dockerfile new file mode 100644 index 0000000..a3f3a6a --- /dev/null +++ b/airflow/Dockerfile @@ -0,0 +1,46 @@ +# +# Dockerfile for airflow +# + +FROM python:3.7-alpine + +ENV AIRFLOW_VERSION=1.10.5 +ENV AIRFLOW_EXTRAS=async,all_dbs,celery,crypto,devel_hadoop,jdbc,ldap,password,redis,s3,samba,slack,ssh,statsd +ENV AIRFLOW_HOME=/opt/airflow +ENV AIRFLOW_CONFIG=airflow.cfg + +ARG FERNET_KEY=4XHGZH0dZ40iOv6z5cyfrXVg5qg3s_d06A7BFfbSsqA= +ENV FERNET_KEY=${FERNET_KEY} + +RUN set -xe \ + && apk add --no-cache \ + build-base \ + cyrus-sasl-dev \ + freetds \ + freetds-dev \ + krb5-dev \ + libffi-dev \ + mariadb-dev \ + postgresql-dev \ + python3-dev \ + && pip install cython numpy \ + && pip install apache-airflow[${AIRFLOW_EXTRAS}]==${AIRFLOW_VERSION} \ + && pip install "websocket-client<0.55.0,>=0.35" \ + && apk del \ + build-base \ + cyrus-sasl-dev \ + freetds-dev \ + krb5-dev \ + libffi-dev \ + mariadb-dev \ + postgresql-dev \ + python3-dev \ + && rm -rf /root/.cache/pip + +WORKDIR ${AIRFLOW_HOME} +VOLUME ${AIRFLOW_HOME} + +EXPOSE 8080 + +ENTRYPOINT ["airflow"] +CMD ["--help"] diff --git a/airflow/README.md b/airflow/README.md index cede12f..1cceb0a 100644 --- a/airflow/README.md +++ b/airflow/README.md @@ -1,6 +1,8 @@ airflow ======= +![](https://airflow.apache.org/_images/pin_large.png) + ## How It Works ``` @@ -20,13 +22,16 @@ airflow ## Quick Start ```bash -# On Master -$ docker-compose up -d -$ chmod 777 data/airflow/dags -$ docker-compose exec webserver cp -r /usr/local/lib/python3.7/site-packages/airflow/example_dags dags +$ python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())' +4XHGZH0dZ40iOv6z5cyfrXVg5qg3s_d06A7BFfbSsqA= -# On Workers -$ docker-compose up -d -$ chmod 777 data/airflow/dags -$ docker-compose exec worker cp -r /usr/local/lib/python3.7/site-packages/airflow/example_dags dags +$ docker stack deploy -c docker-stack.yaml airflow +$ docker service update --replicas-max-per-node=1 airflow_worker +$ docker service update --replicas 3 airflow_worker + +$ curl http://localhost:8080/ +$ curl http://localhost:5555/ ``` + +> :warning: This docker image was built with a static `FERNET_KEY` environment variable. +> You should set another value to it in `docker-stack.yaml`. diff --git a/airflow/data/airflow.cfg b/airflow/data/airflow.cfg new file mode 100644 index 0000000..5a2d71f --- /dev/null +++ b/airflow/data/airflow.cfg @@ -0,0 +1,825 @@ +# -*- coding: utf-8 -*- +# +# 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. + + +# This is the template for Airflow's default configuration. When Airflow is +# imported, it looks for a configuration file at $AIRFLOW_HOME/airflow.cfg. If +# it doesn't exist, Airflow uses this template to generate it by replacing +# variables in curly braces with their global values from configuration.py. + +# Users should not modify this file; they should customize the generated +# airflow.cfg instead. + + +# ----------------------- TEMPLATE BEGINS HERE ----------------------- + +[core] +# The folder where your airflow pipelines live, most likely a +# subfolder in a code repository +# This path must be absolute +dags_folder = {AIRFLOW_HOME}/dags + +# The folder where airflow should store its log files +# This path must be absolute +base_log_folder = {AIRFLOW_HOME}/logs + +# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search. +# Users must supply an Airflow connection id that provides access to the storage +# location. If remote_logging is set to true, see UPDATING.md for additional +# configuration requirements. +remote_logging = False +remote_log_conn_id = +remote_base_log_folder = +encrypt_s3_logs = False + +# Logging level +logging_level = INFO +fab_logging_level = WARN + +# Logging class +# Specify the class that will specify the logging configuration +# This class has to be on the python classpath +# logging_config_class = my.path.default_local_settings.LOGGING_CONFIG +logging_config_class = + +# Log format +# Colour the logs when the controlling terminal is a TTY. +colored_console_log = True +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 +colored_formatter_class = airflow.utils.log.colored_log.CustomTTYColoredFormatter + +log_format = [%%(asctime)s] {{%%(filename)s:%%(lineno)d}} %%(levelname)s - %%(message)s +simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s + +# Specify prefix pattern like mentioned below with stream handler TaskHandlerWithCustomFormatter +# task_log_prefix_template = {{ti.dag_id}}-{{ti.task_id}}-{{execution_date}}-{{try_number}} +task_log_prefix_template = + +# Log filename format +log_filename_template = {{{{ ti.dag_id }}}}/{{{{ ti.task_id }}}}/{{{{ ts }}}}/{{{{ try_number }}}}.log +log_processor_filename_template = {{{{ filename }}}}.log +dag_processor_manager_log_location = {AIRFLOW_HOME}/logs/dag_processor_manager/dag_processor_manager.log + +# Hostname by providing a path to a callable, which will resolve the hostname +# The format is "package:function". For example, +# default value "socket:getfqdn" means that result from getfqdn() of "socket" package will be used as hostname +# 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" +hostname_callable = socket:getfqdn + +# Default timezone in case supplied date times are naive +# can be utc (default), system, or any IANA timezone string (e.g. Europe/Amsterdam) +default_timezone = utc + +# The executor class that airflow should use. Choices include +# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor, KubernetesExecutor +executor = CeleryExecutor + +# The SqlAlchemy connection string to the metadata database. +# SqlAlchemy supports many different database engine, more information +# their website +sql_alchemy_conn = postgresql+psycopg2://airflow:airflow@postgres:5432/airflow + +# The encoding for the databases +sql_engine_encoding = utf-8 + +# If SqlAlchemy should pool database connections. +sql_alchemy_pool_enabled = True + +# The SqlAlchemy pool size is the maximum number of database connections +# in the pool. 0 indicates no limit. +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. +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. +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/13/core/pooling.html#disconnect-handling-pessimistic +sql_alchemy_pool_pre_ping = True + +# The schema to use for the metadata database +# SqlAlchemy supports databases with the concept of multiple schemas. +sql_alchemy_schema = + +# The amount of parallelism as a setting to the executor. This defines +# the max number of task instances that should run simultaneously +# on this airflow installation +parallelism = 32 + +# The number of task instances allowed to run concurrently by the scheduler +dag_concurrency = 16 + +# Are DAGs paused by default at creation +dags_are_paused_at_creation = True + +# The maximum number of active DAG runs per DAG +max_active_runs_per_dag = 16 + +# Whether to load the examples that ship with Airflow. It's good to +# get started, but you probably want to set this to False in a production +# environment +load_examples = False + +# Where your Airflow plugins are stored +plugins_folder = {AIRFLOW_HOME}/plugins + +# Secret key to save connection passwords in the db +fernet_key = {FERNET_KEY} + +# Whether to disable pickling dags +donot_pickle = True + +# How long before timing out a python file import +dagbag_import_timeout = 30 + +# How long before timing out a DagFileProcessor, which processes a dag file +dag_file_processor_timeout = 50 + +# The class to use for running task instances in a subprocess +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 +default_impersonation = + +# What security module to use (for example kerberos): +security = + +# If set to False enables some unsecure features like Charts and Ad Hoc Queries. +# In 2.0 will default to True. +secure_mode = False + +# Turn unit test mode on (overwrites many configuration options with test +# values at runtime) +unit_test_mode = False + +# Name of handler to read task instance logs. +# Default to use task handler. +task_log_reader = task + +# Whether to enable pickling for xcom (note that this is insecure and allows for +# RCE exploits). This will be deprecated in Airflow 2.0 (be forced to False). +enable_xcom_pickling = True + +# 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 +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. +dag_run_conf_overrides_params = False + +# Worker initialisation check to validate Metadata Database connection +worker_precheck = False + +# When discovering DAGs, ignore any files that don't contain the strings `DAG` and `airflow`. +dag_discovery_safe_mode = True + +# The number of retries each task is going to have by default. Can be overridden at dag or task level. +default_task_retries = 0 + + +[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 +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/... +endpoint_url = http://localhost:8080 + +[api] +# How to authenticate users of the API +auth_backend = airflow.api.auth.backend.default + +[lineage] +# what lineage backend to use +backend = + +[atlas] +sasl_enabled = False +host = +port = 21000 +username = +password = + +[operators] +# The default owner assigned to each new operator, unless +# provided explicitly or passed via `default_args` +default_owner = airflow +default_cpus = 1 +default_ram = 512 +default_disk = 512 +default_gpus = 0 + +[hive] +# Default mapreduce queue for HiveOperator tasks +default_hive_mapred_queue = +# Template for mapred_job_name in HiveOperator, supports the following named parameters: +# hostname, dag_id, task_id, execution_date +mapred_job_name_template = Airflow HiveOperator task for {{hostname}}.{{dag_id}}.{{task_id}}.{{execution_date}} + +[webserver] +# The base url of your website as airflow cannot guess what domain or +# cname you are using. This is used in automated emails that +# airflow sends to point links to the right web server +base_url = http://localhost:8080 + +# The ip specified when starting the web server +web_server_host = 0.0.0.0 + +# The port on which to run the web server +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. +web_server_ssl_cert = +web_server_ssl_key = + +# Number of seconds the webserver waits before killing gunicorn master that doesn't respond +web_server_master_timeout = 120 + +# Number of seconds the gunicorn webserver waits before timing out on a worker +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. +worker_refresh_batch_size = 1 + +# Number of seconds to wait before refreshing a batch of workers. +worker_refresh_interval = 30 + +# Secret key used to run your flask app +# It should be as random as possible +secret_key = {SECRET_KEY} + +# Number of workers to run the Gunicorn web server +workers = 4 + +# The worker class gunicorn should use. Choices include +# sync (default), eventlet, gevent +worker_class = sync + +# Log files for the gunicorn webserver. '-' means log to stderr. +access_logfile = - +error_logfile = - + +# Expose the configuration file in the web server +expose_config = False + +# Default DAG view. Valid values are: +# tree, graph, duration, gantt, landing_times +dag_default_view = tree + +# Default DAG orientation. Valid values are: +# LR (Left->Right), TB (Top->Bottom), RL (Right->Left), BT (Bottom->Top) +dag_orientation = LR + +# Puts the webserver in demonstration mode; blurs the names of Operators for +# privacy. +demo_mode = False + +# The amount of time (in secs) webserver will wait for initial handshake +# while fetching logs from other worker machine +log_fetch_timeout_sec = 5 + +# By default, the webserver shows paused DAGs. Flip this to hide paused +# DAGs by default +hide_paused_dags_by_default = False + +# Consistent page size across all listing views in the UI +page_size = 100 + +# Define the color of navigation bar +navbar_color = #007A87 + +# Default dagrun to show in UI +default_dag_run_display_number = 25 + +# Enable werkzeug `ProxyFix` middleware +enable_proxy_fix = False + +# Set secure flag on session cookie +cookie_secure = False + +# Set samesite policy on session cookie +cookie_samesite = + +# Default setting for wrap toggle on DAG code and TI log views. +default_wrap = False + +# Send anonymous user activity to your analytics tool +# analytics_tool = # choose from google_analytics, segment, or metarouter +# analytics_id = XXXXXXXXXXX + +[email] +email_backend = airflow.utils.email.send_email_smtp + + +[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 +smtp_host = localhost +smtp_starttls = True +smtp_ssl = False +# Uncomment and set the user/pass settings if you want to use SMTP AUTH +# smtp_user = airflow +# smtp_password = airflow +smtp_port = 25 +smtp_mail_from = airflow@example.com + +[sentry] +# Sentry (https://docs.sentry.io) integration +sentry_dsn = + + +[celery] +# This section only applies if you are using the CeleryExecutor in +# [core] section above + +# The app name that will be used by celery +celery_app_name = airflow.executors.celery_executor + +# The concurrency that will be used when starting workers with the +# "airflow worker" command. This defines the number of task instances that +# a worker will take, so size up your workers based on the resources on +# your worker box and the nature of your tasks +worker_concurrency = 16 + +# The maximum and minimum concurrency that will be used when starting workers with the +# "airflow worker" command (always keep minimum processes, but grow to maximum if necessary). +# Note the value should be "max_concurrency,min_concurrency" +# Pick these numbers based on resources on worker box and the nature of the task. +# If autoscale option is available, worker_concurrency will be ignored. +# http://docs.celeryproject.org/en/latest/reference/celery.bin.worker.html#cmdoption-celery-worker-autoscale +# worker_autoscale = 16,12 + +# 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. +worker_log_server_port = 8793 + +# The Celery broker URL. Celery supports RabbitMQ, Redis and experimentally +# a sqlalchemy database. Refer to the Celery documentation for more +# information. +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#broker-settings +broker_url = redis://redis:6379/1 + +# The Celery result_backend. When a job finishes, it needs to update the +# metadata of the job. Therefore it will post a message on a message bus, +# or insert it into a database (depending of the backend) +# This status is used by the scheduler to update the state of the task +# The use of a database is highly recommended +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-result-backend-settings +result_backend = db+postgresql://airflow:airflow@postgres/airflow + +# Celery Flower is a sweet UI for Celery. Airflow has a shortcut to start +# it `airflow flower`. This defines the IP that Celery Flower runs on +flower_host = 0.0.0.0 + +# The root URL for Flower +# Ex: flower_url_prefix = /flower +flower_url_prefix = + +# This defines the port that Celery Flower runs on +flower_port = 5555 + +# Securing Flower with Basic Authentication +# Accepts user:password pairs separated by a comma +# Example: flower_basic_auth = user1:password1,user2:password2 +flower_basic_auth = + +# Default queue that tasks get assigned to and that worker listen on. +default_queue = default + +# How many processes CeleryExecutor uses to sync task state. +# 0 means to use max(1, number of cores - 1) processes. +sync_parallelism = 0 + +# Import path for celery configuration options +celery_config_options = airflow.config_templates.default_celery.DEFAULT_CELERY_CONFIG + +# In case of using SSL +ssl_active = False +ssl_key = +ssl_cert = +ssl_cacert = + +# Celery Pool implementation. +# Choices include: prefork (default), eventlet, gevent or solo. +# See: +# https://docs.celeryproject.org/en/latest/userguide/workers.html#concurrency +# https://docs.celeryproject.org/en/latest/userguide/concurrency/eventlet.html +pool = prefork + +[celery_broker_transport_options] +# This section is for specifying options which can be passed to the +# underlying celery broker transport. See: +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#std:setting-broker_transport_options + +# The visibility timeout defines the number of seconds to wait for the worker +# to acknowledge the task before the message is redelivered to another worker. +# Make sure to increase the visibility timeout to match the time of the longest +# ETA you're planning to use. +# +# visibility_timeout is only supported for Redis and SQS celery brokers. +# See: +# http://docs.celeryproject.org/en/master/userguide/configuration.html#std:setting-broker_transport_options +# +#visibility_timeout = 21600 + +[dask] +# This section only applies if you are using the DaskExecutor in +# [core] section above + +# The IP address and port of the Dask cluster's scheduler. +cluster_address = 127.0.0.1:8786 +# TLS/ SSL settings to access a secured Dask scheduler. +tls_ca = +tls_cert = +tls_key = + + +[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). +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). +scheduler_heartbeat_sec = 5 + +# The number of times to try to schedule each DAG file +# -1 indicates unlimited number +num_runs = -1 + +# The number of seconds to wait between consecutive DAG file processing +processor_poll_interval = 1 + +# after how much time (seconds) a new DAGs should be picked up from the filesystem +min_file_process_interval = 0 + +# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes. +dag_dir_list_interval = 300 + +# How often should stats be printed to the logs +print_stats_interval = 30 + +# 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 +scheduler_health_check_threshold = 30 + +child_process_log_directory = {AIRFLOW_HOME}/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. +scheduler_zombie_task_threshold = 300 + +# 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) +catchup_by_default = True + +# This changes the batch size of queries in the scheduling main loop. +# If this is too high, SQL query performance may be impacted by one +# or more of the following: +# - reversion to full table scan +# - complexity of query predicate +# - excessive locking +# +# Additionally, you may hit the maximum allowable query length for your db. +# +# Set this to 0 for no limit (not advised) +max_tis_per_query = 512 + +# Statsd (https://github.com/etsy/statsd) integration settings +statsd_on = False +statsd_host = localhost +statsd_port = 8125 +statsd_prefix = airflow + +# If you want to avoid send all the available metrics to StatsD, +# you can configure an allow list of prefixes to send only the metrics that +# start with the elements of the list (e.g: scheduler,executor,dagrun) +statsd_allow_list = + +# The scheduler can run multiple threads in parallel to schedule dags. +# This defines how many threads will run. +max_threads = 2 + +authenticate = False + +# 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. +use_job_schedule = True + +[ldap] +# set this to ldaps://: +uri = +user_filter = objectClass=* +user_name_attr = uid +group_member_attr = memberOf +superuser_filter = +data_profiler_filter = +bind_user = cn=Manager,dc=example,dc=com +bind_password = insecure +basedn = dc=example,dc=com +cacert = /etc/ca/ldap_ca.crt +search_scope = LEVEL + +# This setting allows the use of LDAP servers that either return a +# broken schema, or do not return a schema. +ignore_malformed_schema = False + +[kerberos] +ccache = /tmp/airflow_krb5_ccache +# gets augmented with fqdn +principal = airflow +reinit_frequency = 3600 +kinit_path = kinit +keytab = airflow.keytab + + +[github_enterprise] +api_rev = v3 + +[admin] +# UI to hide sensitive variable fields when set to True +hide_sensitive_variable_fields = True + +[elasticsearch] +# Elasticsearch host +host = +# Format of the log_id, which is used to query for a given tasks logs +log_id_template = {{dag_id}}-{{task_id}}-{{execution_date}}-{{try_number}} +# Used to mark the end of a log stream for a task +end_of_log_mark = end_of_log +# Qualified URL for an elasticsearch frontend (like Kibana) with a template argument for log_id +# Code will construct log_id using the log_id template from the argument above. +# NOTE: The code will prefix the https:// automatically, don't include that here. +frontend = +# Write the task logs to the stdout of the worker, rather than the default files +write_stdout = False +# Instead of the default log formatter, write the log lines as JSON +json_format = False +# Log fields to also attach to the json output, if enabled +json_fields = asctime, filename, lineno, levelname, message + +[elasticsearch_configs] + +use_ssl = False +verify_certs = True + +[kubernetes] +# The repository, tag and imagePullPolicy of the Kubernetes Image for the Worker to Run +worker_container_repository = +worker_container_tag = +worker_container_image_pull_policy = IfNotPresent + +# If True (default), worker pods will be deleted upon termination +delete_worker_pods = True + +# Number of Kubernetes Worker Pod creation calls per scheduler loop +worker_pods_creation_batch_size = 1 + +# The Kubernetes namespace where airflow workers should be created. Defaults to `default` +namespace = default + +# The name of the Kubernetes ConfigMap Containing the Airflow Configuration (this file) +airflow_configmap = + +# For docker image already contains DAGs, this is set to `True`, and the worker will search for dags in dags_folder, +# otherwise use git sync or dags volume claim to mount DAGs +dags_in_image = False + +# For either git sync or volume mounted DAGs, the worker will look in this subpath for DAGs +dags_volume_subpath = + +# For DAGs mounted via a volume claim (mutually exclusive with git-sync and host path) +dags_volume_claim = + +# For volume mounted logs, the worker will look in this subpath for logs +logs_volume_subpath = + +# A shared volume claim for the logs +logs_volume_claim = + +# For DAGs mounted via a hostPath volume (mutually exclusive with volume claim and git-sync) +# Useful in local environment, discouraged in production +dags_volume_host = + +# A hostPath volume for the logs +# Useful in local environment, discouraged in production +logs_volume_host = + +# A list of configMapsRefs to envFrom. If more than one configMap is +# specified, provide a comma separated list: configmap_a,configmap_b +env_from_configmap_ref = + +# A list of secretRefs to envFrom. If more than one secret is +# specified, provide a comma separated list: secret_a,secret_b +env_from_secret_ref = + +# Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim) +git_repo = +git_branch = +git_subpath = +# Use git_user and git_password for user authentication or git_ssh_key_secret_name and git_ssh_key_secret_key +# for SSH authentication +git_user = +git_password = +git_sync_root = /git +git_sync_dest = repo +# Mount point of the volume if git-sync is being used. +# i.e. {AIRFLOW_HOME}/dags +git_dags_folder_mount_point = + +# To get Git-sync SSH authentication set up follow this format +# +# airflow-secrets.yaml: +# --- +# apiVersion: v1 +# kind: Secret +# metadata: +# name: airflow-secrets +# data: +# # key needs to be gitSshKey +# gitSshKey: +# --- +# airflow-configmap.yaml: +# apiVersion: v1 +# kind: ConfigMap +# metadata: +# name: airflow-configmap +# data: +# known_hosts: | +# github.com ssh-rsa <...> +# airflow.cfg: | +# ... +# +# git_ssh_key_secret_name = airflow-secrets +# git_ssh_known_hosts_configmap_name = airflow-configmap +git_ssh_key_secret_name = +git_ssh_known_hosts_configmap_name = + +# To give the git_sync init container credentials via a secret, create a secret +# with two fields: GIT_SYNC_USERNAME and GIT_SYNC_PASSWORD (example below) and +# add `git_sync_credentials_secret = ` to your airflow config under the kubernetes section +# +# Secret Example: +# apiVersion: v1 +# kind: Secret +# metadata: +# name: git-credentials +# data: +# GIT_SYNC_USERNAME: +# GIT_SYNC_PASSWORD: +git_sync_credentials_secret = + +# For cloning DAGs from git repositories into volumes: https://github.com/kubernetes/git-sync +git_sync_container_repository = k8s.gcr.io/git-sync +git_sync_container_tag = v3.1.1 +git_sync_init_container_name = git-sync-clone +git_sync_run_as_user = 65533 + +# The name of the Kubernetes service account to be associated with airflow workers, if any. +# Service accounts are required for workers that require access to secrets or cluster resources. +# See the Kubernetes RBAC documentation for more: +# https://kubernetes.io/docs/admin/authorization/rbac/ +worker_service_account_name = + +# Any image pull secrets to be given to worker pods, If more than one secret is +# required, provide a comma separated list: secret_a,secret_b +image_pull_secrets = + +# GCP Service Account Keys to be provided to tasks run on Kubernetes Executors +# Should be supplied in the format: key-name-1:key-path-1,key-name-2:key-path-2 +gcp_service_account_keys = + +# Use the service account kubernetes gives to pods to connect to kubernetes cluster. +# It's intended for clients that expect to be running inside a pod running on kubernetes. +# It will raise an exception if called from a process not running in a kubernetes environment. +in_cluster = True + +# When running with in_cluster=False change the default cluster_context or config_file +# options to Kubernetes client. Leave blank these to use default behaviour like `kubectl` has. +# cluster_context = +# config_file = + + +# Affinity configuration as a single line formatted JSON object. +# See the affinity model for top-level key names (e.g. `nodeAffinity`, etc.): +# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#affinity-v1-core +affinity = + +# A list of toleration objects as a single line formatted JSON array +# See: +# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#toleration-v1-core +tolerations = + +# **kwargs parameters to pass while calling a kubernetes client core_v1_api methods from Kubernetes Executor +# provided as a single line formatted JSON dictionary string. +# List of supported params in **kwargs are similar for all core_v1_apis, hence a single config variable for all apis +# See: +# https://raw.githubusercontent.com/kubernetes-client/python/master/kubernetes/client/apis/core_v1_api.py +kube_client_request_args = + +# Worker pods security context options +# See: +# https://kubernetes.io/docs/tasks/configure-pod-container/security-context/ + +# Specifies the uid to run the first process of the worker pods containers as +run_as_user = + +# Specifies a gid to associate with all containers in the worker pods +# if using a git_ssh_key_secret_name use an fs_group +# that allows for the key to be read, e.g. 65533 +fs_group = + +# Annotations configuration as a single line formatted JSON object. +# See the naming convention in: +# https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/ +worker_annotations = + + +[kubernetes_node_selectors] +# The Key-value pairs to be given to worker pods. +# The worker pods will be scheduled to the nodes of the specified key-value pairs. +# Should be supplied in the format: key = value + +[kubernetes_environment_variables] +# The scheduler sets the following environment variables into your workers. You may define as +# many environment variables as needed and the kubernetes launcher will set them in the launched workers. +# Environment variables in this section are defined as follows +# = +# +# For example if you wanted to set an environment variable with value `prod` and key +# `ENVIRONMENT` you would follow the following format: +# ENVIRONMENT = prod +# +# Additionally you may override worker airflow settings with the AIRFLOW__
__ +# formatting as supported by airflow normally. + +[kubernetes_secrets] +# The scheduler mounts the following secrets into your workers as they are launched by the +# scheduler. You may define as many secrets as needed and the kubernetes launcher will parse the +# defined secrets and mount them as secret environment variables in the launched workers. +# Secrets in this section are defined as follows +# = = +# +# For example if you wanted to mount a kubernetes secret key named `postgres_password` from the +# kubernetes secret object `airflow-secret` as the environment variable `POSTGRES_PASSWORD` into +# your workers you would follow the following format: +# POSTGRES_PASSWORD = airflow-secret=postgres_credentials +# +# Additionally you may override worker airflow settings with the AIRFLOW__
__ +# formatting as supported by airflow normally. + +[kubernetes_labels] +# The Key-value pairs to be given to worker pods. +# The worker pods will be given these static labels, as well as some additional dynamic labels +# to identify the task. +# Should be supplied in the format: key = value diff --git a/airflow/data/dags/.gitkeep b/airflow/data/dags/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/airflow/data/dags/__init__.py b/airflow/data/dags/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/airflow/data/dags/tutorial.py b/airflow/data/dags/tutorial.py new file mode 100644 index 0000000..e994c1c --- /dev/null +++ b/airflow/data/dags/tutorial.py @@ -0,0 +1,104 @@ +# -*- coding: utf-8 -*- +# +# 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. + +""" +### Tutorial Documentation +Documentation that goes along with the Airflow tutorial located +[here](https://airflow.apache.org/tutorial.html) +""" +from datetime import timedelta + +import airflow +from airflow import DAG +from airflow.operators.bash_operator import BashOperator + +# These args will get passed on to each operator +# You can override them on a per-task basis during operator initialization +default_args = { + 'owner': 'Airflow', + 'depends_on_past': False, + 'start_date': airflow.utils.dates.days_ago(2), + 'email': ['airflow@example.com'], + 'email_on_failure': False, + 'email_on_retry': False, + 'retries': 1, + 'retry_delay': timedelta(minutes=5), + # 'queue': 'bash_queue', + # 'pool': 'backfill', + # 'priority_weight': 10, + # 'end_date': datetime(2016, 1, 1), + # 'wait_for_downstream': False, + # 'dag': dag, + # 'sla': timedelta(hours=2), + # 'execution_timeout': timedelta(seconds=300), + # 'on_failure_callback': some_function, + # 'on_success_callback': some_other_function, + # 'on_retry_callback': another_function, + # 'sla_miss_callback': yet_another_function, + # 'trigger_rule': 'all_success' +} + +dag = DAG( + 'tutorial', + default_args=default_args, + description='A simple tutorial DAG', + schedule_interval=timedelta(days=1), +) + +# t1, t2 and t3 are examples of tasks created by instantiating operators +t1 = BashOperator( + task_id='print_date', + bash_command='date', + dag=dag, +) + +t1.doc_md = """\ +#### Task Documentation +You can document your task using the attributes `doc_md` (markdown), +`doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets +rendered in the UI's Task Instance Details page. +![img](http://montcs.bloomu.edu/~bobmon/Semesters/2012-01/491/import%20soul.png) +""" + +dag.doc_md = __doc__ + +t2 = BashOperator( + task_id='sleep', + depends_on_past=False, + bash_command='sleep 5', + dag=dag, +) + +templated_command = """ +{% for i in range(5) %} + echo "{{ ds }}" + echo "{{ macros.ds_add(ds, 7)}}" + echo "{{ params.my_param }}" +{% endfor %} +""" + +t3 = BashOperator( + task_id='templated', + depends_on_past=False, + bash_command=templated_command, + params={'my_param': 'Parameter I passed in'}, + dag=dag, +) + +t1 >> [t2, t3] diff --git a/airflow/data/default_airflow.cfg b/airflow/data/default_airflow.cfg new file mode 100644 index 0000000..ae46feb --- /dev/null +++ b/airflow/data/default_airflow.cfg @@ -0,0 +1,825 @@ +# -*- coding: utf-8 -*- +# +# 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. + + +# This is the template for Airflow's default configuration. When Airflow is +# imported, it looks for a configuration file at $AIRFLOW_HOME/airflow.cfg. If +# it doesn't exist, Airflow uses this template to generate it by replacing +# variables in curly braces with their global values from configuration.py. + +# Users should not modify this file; they should customize the generated +# airflow.cfg instead. + + +# ----------------------- TEMPLATE BEGINS HERE ----------------------- + +[core] +# The folder where your airflow pipelines live, most likely a +# subfolder in a code repository +# This path must be absolute +dags_folder = {AIRFLOW_HOME}/dags + +# The folder where airflow should store its log files +# This path must be absolute +base_log_folder = {AIRFLOW_HOME}/logs + +# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search. +# Users must supply an Airflow connection id that provides access to the storage +# location. If remote_logging is set to true, see UPDATING.md for additional +# configuration requirements. +remote_logging = False +remote_log_conn_id = +remote_base_log_folder = +encrypt_s3_logs = False + +# Logging level +logging_level = INFO +fab_logging_level = WARN + +# Logging class +# Specify the class that will specify the logging configuration +# This class has to be on the python classpath +# logging_config_class = my.path.default_local_settings.LOGGING_CONFIG +logging_config_class = + +# Log format +# Colour the logs when the controlling terminal is a TTY. +colored_console_log = True +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 +colored_formatter_class = airflow.utils.log.colored_log.CustomTTYColoredFormatter + +log_format = [%%(asctime)s] {{%%(filename)s:%%(lineno)d}} %%(levelname)s - %%(message)s +simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s + +# Specify prefix pattern like mentioned below with stream handler TaskHandlerWithCustomFormatter +# task_log_prefix_template = {{ti.dag_id}}-{{ti.task_id}}-{{execution_date}}-{{try_number}} +task_log_prefix_template = + +# Log filename format +log_filename_template = {{{{ ti.dag_id }}}}/{{{{ ti.task_id }}}}/{{{{ ts }}}}/{{{{ try_number }}}}.log +log_processor_filename_template = {{{{ filename }}}}.log +dag_processor_manager_log_location = {AIRFLOW_HOME}/logs/dag_processor_manager/dag_processor_manager.log + +# Hostname by providing a path to a callable, which will resolve the hostname +# The format is "package:function". For example, +# default value "socket:getfqdn" means that result from getfqdn() of "socket" package will be used as hostname +# 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" +hostname_callable = socket:getfqdn + +# Default timezone in case supplied date times are naive +# can be utc (default), system, or any IANA timezone string (e.g. Europe/Amsterdam) +default_timezone = utc + +# The executor class that airflow should use. Choices include +# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor, KubernetesExecutor +executor = SequentialExecutor + +# The SqlAlchemy connection string to the metadata database. +# SqlAlchemy supports many different database engine, more information +# their website +sql_alchemy_conn = sqlite:///{AIRFLOW_HOME}/airflow.db + +# The encoding for the databases +sql_engine_encoding = utf-8 + +# If SqlAlchemy should pool database connections. +sql_alchemy_pool_enabled = True + +# The SqlAlchemy pool size is the maximum number of database connections +# in the pool. 0 indicates no limit. +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. +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. +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/13/core/pooling.html#disconnect-handling-pessimistic +sql_alchemy_pool_pre_ping = True + +# The schema to use for the metadata database +# SqlAlchemy supports databases with the concept of multiple schemas. +sql_alchemy_schema = + +# The amount of parallelism as a setting to the executor. This defines +# the max number of task instances that should run simultaneously +# on this airflow installation +parallelism = 32 + +# The number of task instances allowed to run concurrently by the scheduler +dag_concurrency = 16 + +# Are DAGs paused by default at creation +dags_are_paused_at_creation = True + +# The maximum number of active DAG runs per DAG +max_active_runs_per_dag = 16 + +# Whether to load the examples that ship with Airflow. It's good to +# get started, but you probably want to set this to False in a production +# environment +load_examples = True + +# Where your Airflow plugins are stored +plugins_folder = {AIRFLOW_HOME}/plugins + +# Secret key to save connection passwords in the db +fernet_key = {FERNET_KEY} + +# Whether to disable pickling dags +donot_pickle = True + +# How long before timing out a python file import +dagbag_import_timeout = 30 + +# How long before timing out a DagFileProcessor, which processes a dag file +dag_file_processor_timeout = 50 + +# The class to use for running task instances in a subprocess +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 +default_impersonation = + +# What security module to use (for example kerberos): +security = + +# If set to False enables some unsecure features like Charts and Ad Hoc Queries. +# In 2.0 will default to True. +secure_mode = False + +# Turn unit test mode on (overwrites many configuration options with test +# values at runtime) +unit_test_mode = False + +# Name of handler to read task instance logs. +# Default to use task handler. +task_log_reader = task + +# Whether to enable pickling for xcom (note that this is insecure and allows for +# RCE exploits). This will be deprecated in Airflow 2.0 (be forced to False). +enable_xcom_pickling = True + +# 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 +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. +dag_run_conf_overrides_params = False + +# Worker initialisation check to validate Metadata Database connection +worker_precheck = False + +# When discovering DAGs, ignore any files that don't contain the strings `DAG` and `airflow`. +dag_discovery_safe_mode = True + +# The number of retries each task is going to have by default. Can be overridden at dag or task level. +default_task_retries = 0 + + +[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 +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/... +endpoint_url = http://localhost:8080 + +[api] +# How to authenticate users of the API +auth_backend = airflow.api.auth.backend.default + +[lineage] +# what lineage backend to use +backend = + +[atlas] +sasl_enabled = False +host = +port = 21000 +username = +password = + +[operators] +# The default owner assigned to each new operator, unless +# provided explicitly or passed via `default_args` +default_owner = airflow +default_cpus = 1 +default_ram = 512 +default_disk = 512 +default_gpus = 0 + +[hive] +# Default mapreduce queue for HiveOperator tasks +default_hive_mapred_queue = +# Template for mapred_job_name in HiveOperator, supports the following named parameters: +# hostname, dag_id, task_id, execution_date +mapred_job_name_template = Airflow HiveOperator task for {{hostname}}.{{dag_id}}.{{task_id}}.{{execution_date}} + +[webserver] +# The base url of your website as airflow cannot guess what domain or +# cname you are using. This is used in automated emails that +# airflow sends to point links to the right web server +base_url = http://localhost:8080 + +# The ip specified when starting the web server +web_server_host = 0.0.0.0 + +# The port on which to run the web server +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. +web_server_ssl_cert = +web_server_ssl_key = + +# Number of seconds the webserver waits before killing gunicorn master that doesn't respond +web_server_master_timeout = 120 + +# Number of seconds the gunicorn webserver waits before timing out on a worker +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. +worker_refresh_batch_size = 1 + +# Number of seconds to wait before refreshing a batch of workers. +worker_refresh_interval = 30 + +# Secret key used to run your flask app +# It should be as random as possible +secret_key = {SECRET_KEY} + +# Number of workers to run the Gunicorn web server +workers = 4 + +# The worker class gunicorn should use. Choices include +# sync (default), eventlet, gevent +worker_class = sync + +# Log files for the gunicorn webserver. '-' means log to stderr. +access_logfile = - +error_logfile = - + +# Expose the configuration file in the web server +expose_config = False + +# Default DAG view. Valid values are: +# tree, graph, duration, gantt, landing_times +dag_default_view = tree + +# Default DAG orientation. Valid values are: +# LR (Left->Right), TB (Top->Bottom), RL (Right->Left), BT (Bottom->Top) +dag_orientation = LR + +# Puts the webserver in demonstration mode; blurs the names of Operators for +# privacy. +demo_mode = False + +# The amount of time (in secs) webserver will wait for initial handshake +# while fetching logs from other worker machine +log_fetch_timeout_sec = 5 + +# By default, the webserver shows paused DAGs. Flip this to hide paused +# DAGs by default +hide_paused_dags_by_default = False + +# Consistent page size across all listing views in the UI +page_size = 100 + +# Define the color of navigation bar +navbar_color = #007A87 + +# Default dagrun to show in UI +default_dag_run_display_number = 25 + +# Enable werkzeug `ProxyFix` middleware +enable_proxy_fix = False + +# Set secure flag on session cookie +cookie_secure = False + +# Set samesite policy on session cookie +cookie_samesite = + +# Default setting for wrap toggle on DAG code and TI log views. +default_wrap = False + +# Send anonymous user activity to your analytics tool +# analytics_tool = # choose from google_analytics, segment, or metarouter +# analytics_id = XXXXXXXXXXX + +[email] +email_backend = airflow.utils.email.send_email_smtp + + +[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 +smtp_host = localhost +smtp_starttls = True +smtp_ssl = False +# Uncomment and set the user/pass settings if you want to use SMTP AUTH +# smtp_user = airflow +# smtp_password = airflow +smtp_port = 25 +smtp_mail_from = airflow@example.com + +[sentry] +# Sentry (https://docs.sentry.io) integration +sentry_dsn = + + +[celery] +# This section only applies if you are using the CeleryExecutor in +# [core] section above + +# The app name that will be used by celery +celery_app_name = airflow.executors.celery_executor + +# The concurrency that will be used when starting workers with the +# "airflow worker" command. This defines the number of task instances that +# a worker will take, so size up your workers based on the resources on +# your worker box and the nature of your tasks +worker_concurrency = 16 + +# The maximum and minimum concurrency that will be used when starting workers with the +# "airflow worker" command (always keep minimum processes, but grow to maximum if necessary). +# Note the value should be "max_concurrency,min_concurrency" +# Pick these numbers based on resources on worker box and the nature of the task. +# If autoscale option is available, worker_concurrency will be ignored. +# http://docs.celeryproject.org/en/latest/reference/celery.bin.worker.html#cmdoption-celery-worker-autoscale +# worker_autoscale = 16,12 + +# 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. +worker_log_server_port = 8793 + +# The Celery broker URL. Celery supports RabbitMQ, Redis and experimentally +# a sqlalchemy database. Refer to the Celery documentation for more +# information. +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#broker-settings +broker_url = sqla+mysql://airflow:airflow@localhost:3306/airflow + +# The Celery result_backend. When a job finishes, it needs to update the +# metadata of the job. Therefore it will post a message on a message bus, +# or insert it into a database (depending of the backend) +# This status is used by the scheduler to update the state of the task +# The use of a database is highly recommended +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-result-backend-settings +result_backend = db+mysql://airflow:airflow@localhost:3306/airflow + +# Celery Flower is a sweet UI for Celery. Airflow has a shortcut to start +# it `airflow flower`. This defines the IP that Celery Flower runs on +flower_host = 0.0.0.0 + +# The root URL for Flower +# Ex: flower_url_prefix = /flower +flower_url_prefix = + +# This defines the port that Celery Flower runs on +flower_port = 5555 + +# Securing Flower with Basic Authentication +# Accepts user:password pairs separated by a comma +# Example: flower_basic_auth = user1:password1,user2:password2 +flower_basic_auth = + +# Default queue that tasks get assigned to and that worker listen on. +default_queue = default + +# How many processes CeleryExecutor uses to sync task state. +# 0 means to use max(1, number of cores - 1) processes. +sync_parallelism = 0 + +# Import path for celery configuration options +celery_config_options = airflow.config_templates.default_celery.DEFAULT_CELERY_CONFIG + +# In case of using SSL +ssl_active = False +ssl_key = +ssl_cert = +ssl_cacert = + +# Celery Pool implementation. +# Choices include: prefork (default), eventlet, gevent or solo. +# See: +# https://docs.celeryproject.org/en/latest/userguide/workers.html#concurrency +# https://docs.celeryproject.org/en/latest/userguide/concurrency/eventlet.html +pool = prefork + +[celery_broker_transport_options] +# This section is for specifying options which can be passed to the +# underlying celery broker transport. See: +# http://docs.celeryproject.org/en/latest/userguide/configuration.html#std:setting-broker_transport_options + +# The visibility timeout defines the number of seconds to wait for the worker +# to acknowledge the task before the message is redelivered to another worker. +# Make sure to increase the visibility timeout to match the time of the longest +# ETA you're planning to use. +# +# visibility_timeout is only supported for Redis and SQS celery brokers. +# See: +# http://docs.celeryproject.org/en/master/userguide/configuration.html#std:setting-broker_transport_options +# +#visibility_timeout = 21600 + +[dask] +# This section only applies if you are using the DaskExecutor in +# [core] section above + +# The IP address and port of the Dask cluster's scheduler. +cluster_address = 127.0.0.1:8786 +# TLS/ SSL settings to access a secured Dask scheduler. +tls_ca = +tls_cert = +tls_key = + + +[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). +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). +scheduler_heartbeat_sec = 5 + +# The number of times to try to schedule each DAG file +# -1 indicates unlimited number +num_runs = -1 + +# The number of seconds to wait between consecutive DAG file processing +processor_poll_interval = 1 + +# after how much time (seconds) a new DAGs should be picked up from the filesystem +min_file_process_interval = 0 + +# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes. +dag_dir_list_interval = 300 + +# How often should stats be printed to the logs +print_stats_interval = 30 + +# 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 +scheduler_health_check_threshold = 30 + +child_process_log_directory = {AIRFLOW_HOME}/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. +scheduler_zombie_task_threshold = 300 + +# 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) +catchup_by_default = True + +# This changes the batch size of queries in the scheduling main loop. +# If this is too high, SQL query performance may be impacted by one +# or more of the following: +# - reversion to full table scan +# - complexity of query predicate +# - excessive locking +# +# Additionally, you may hit the maximum allowable query length for your db. +# +# Set this to 0 for no limit (not advised) +max_tis_per_query = 512 + +# Statsd (https://github.com/etsy/statsd) integration settings +statsd_on = False +statsd_host = localhost +statsd_port = 8125 +statsd_prefix = airflow + +# If you want to avoid send all the available metrics to StatsD, +# you can configure an allow list of prefixes to send only the metrics that +# start with the elements of the list (e.g: scheduler,executor,dagrun) +statsd_allow_list = + +# The scheduler can run multiple threads in parallel to schedule dags. +# This defines how many threads will run. +max_threads = 2 + +authenticate = False + +# 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. +use_job_schedule = True + +[ldap] +# set this to ldaps://: +uri = +user_filter = objectClass=* +user_name_attr = uid +group_member_attr = memberOf +superuser_filter = +data_profiler_filter = +bind_user = cn=Manager,dc=example,dc=com +bind_password = insecure +basedn = dc=example,dc=com +cacert = /etc/ca/ldap_ca.crt +search_scope = LEVEL + +# This setting allows the use of LDAP servers that either return a +# broken schema, or do not return a schema. +ignore_malformed_schema = False + +[kerberos] +ccache = /tmp/airflow_krb5_ccache +# gets augmented with fqdn +principal = airflow +reinit_frequency = 3600 +kinit_path = kinit +keytab = airflow.keytab + + +[github_enterprise] +api_rev = v3 + +[admin] +# UI to hide sensitive variable fields when set to True +hide_sensitive_variable_fields = True + +[elasticsearch] +# Elasticsearch host +host = +# Format of the log_id, which is used to query for a given tasks logs +log_id_template = {{dag_id}}-{{task_id}}-{{execution_date}}-{{try_number}} +# Used to mark the end of a log stream for a task +end_of_log_mark = end_of_log +# Qualified URL for an elasticsearch frontend (like Kibana) with a template argument for log_id +# Code will construct log_id using the log_id template from the argument above. +# NOTE: The code will prefix the https:// automatically, don't include that here. +frontend = +# Write the task logs to the stdout of the worker, rather than the default files +write_stdout = False +# Instead of the default log formatter, write the log lines as JSON +json_format = False +# Log fields to also attach to the json output, if enabled +json_fields = asctime, filename, lineno, levelname, message + +[elasticsearch_configs] + +use_ssl = False +verify_certs = True + +[kubernetes] +# The repository, tag and imagePullPolicy of the Kubernetes Image for the Worker to Run +worker_container_repository = +worker_container_tag = +worker_container_image_pull_policy = IfNotPresent + +# If True (default), worker pods will be deleted upon termination +delete_worker_pods = True + +# Number of Kubernetes Worker Pod creation calls per scheduler loop +worker_pods_creation_batch_size = 1 + +# The Kubernetes namespace where airflow workers should be created. Defaults to `default` +namespace = default + +# The name of the Kubernetes ConfigMap Containing the Airflow Configuration (this file) +airflow_configmap = + +# For docker image already contains DAGs, this is set to `True`, and the worker will search for dags in dags_folder, +# otherwise use git sync or dags volume claim to mount DAGs +dags_in_image = False + +# For either git sync or volume mounted DAGs, the worker will look in this subpath for DAGs +dags_volume_subpath = + +# For DAGs mounted via a volume claim (mutually exclusive with git-sync and host path) +dags_volume_claim = + +# For volume mounted logs, the worker will look in this subpath for logs +logs_volume_subpath = + +# A shared volume claim for the logs +logs_volume_claim = + +# For DAGs mounted via a hostPath volume (mutually exclusive with volume claim and git-sync) +# Useful in local environment, discouraged in production +dags_volume_host = + +# A hostPath volume for the logs +# Useful in local environment, discouraged in production +logs_volume_host = + +# A list of configMapsRefs to envFrom. If more than one configMap is +# specified, provide a comma separated list: configmap_a,configmap_b +env_from_configmap_ref = + +# A list of secretRefs to envFrom. If more than one secret is +# specified, provide a comma separated list: secret_a,secret_b +env_from_secret_ref = + +# Git credentials and repository for DAGs mounted via Git (mutually exclusive with volume claim) +git_repo = +git_branch = +git_subpath = +# Use git_user and git_password for user authentication or git_ssh_key_secret_name and git_ssh_key_secret_key +# for SSH authentication +git_user = +git_password = +git_sync_root = /git +git_sync_dest = repo +# Mount point of the volume if git-sync is being used. +# i.e. {AIRFLOW_HOME}/dags +git_dags_folder_mount_point = + +# To get Git-sync SSH authentication set up follow this format +# +# airflow-secrets.yaml: +# --- +# apiVersion: v1 +# kind: Secret +# metadata: +# name: airflow-secrets +# data: +# # key needs to be gitSshKey +# gitSshKey: +# --- +# airflow-configmap.yaml: +# apiVersion: v1 +# kind: ConfigMap +# metadata: +# name: airflow-configmap +# data: +# known_hosts: | +# github.com ssh-rsa <...> +# airflow.cfg: | +# ... +# +# git_ssh_key_secret_name = airflow-secrets +# git_ssh_known_hosts_configmap_name = airflow-configmap +git_ssh_key_secret_name = +git_ssh_known_hosts_configmap_name = + +# To give the git_sync init container credentials via a secret, create a secret +# with two fields: GIT_SYNC_USERNAME and GIT_SYNC_PASSWORD (example below) and +# add `git_sync_credentials_secret = ` to your airflow config under the kubernetes section +# +# Secret Example: +# apiVersion: v1 +# kind: Secret +# metadata: +# name: git-credentials +# data: +# GIT_SYNC_USERNAME: +# GIT_SYNC_PASSWORD: +git_sync_credentials_secret = + +# For cloning DAGs from git repositories into volumes: https://github.com/kubernetes/git-sync +git_sync_container_repository = k8s.gcr.io/git-sync +git_sync_container_tag = v3.1.1 +git_sync_init_container_name = git-sync-clone +git_sync_run_as_user = 65533 + +# The name of the Kubernetes service account to be associated with airflow workers, if any. +# Service accounts are required for workers that require access to secrets or cluster resources. +# See the Kubernetes RBAC documentation for more: +# https://kubernetes.io/docs/admin/authorization/rbac/ +worker_service_account_name = + +# Any image pull secrets to be given to worker pods, If more than one secret is +# required, provide a comma separated list: secret_a,secret_b +image_pull_secrets = + +# GCP Service Account Keys to be provided to tasks run on Kubernetes Executors +# Should be supplied in the format: key-name-1:key-path-1,key-name-2:key-path-2 +gcp_service_account_keys = + +# Use the service account kubernetes gives to pods to connect to kubernetes cluster. +# It's intended for clients that expect to be running inside a pod running on kubernetes. +# It will raise an exception if called from a process not running in a kubernetes environment. +in_cluster = True + +# When running with in_cluster=False change the default cluster_context or config_file +# options to Kubernetes client. Leave blank these to use default behaviour like `kubectl` has. +# cluster_context = +# config_file = + + +# Affinity configuration as a single line formatted JSON object. +# See the affinity model for top-level key names (e.g. `nodeAffinity`, etc.): +# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#affinity-v1-core +affinity = + +# A list of toleration objects as a single line formatted JSON array +# See: +# https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.12/#toleration-v1-core +tolerations = + +# **kwargs parameters to pass while calling a kubernetes client core_v1_api methods from Kubernetes Executor +# provided as a single line formatted JSON dictionary string. +# List of supported params in **kwargs are similar for all core_v1_apis, hence a single config variable for all apis +# See: +# https://raw.githubusercontent.com/kubernetes-client/python/master/kubernetes/client/apis/core_v1_api.py +kube_client_request_args = + +# Worker pods security context options +# See: +# https://kubernetes.io/docs/tasks/configure-pod-container/security-context/ + +# Specifies the uid to run the first process of the worker pods containers as +run_as_user = + +# Specifies a gid to associate with all containers in the worker pods +# if using a git_ssh_key_secret_name use an fs_group +# that allows for the key to be read, e.g. 65533 +fs_group = + +# Annotations configuration as a single line formatted JSON object. +# See the naming convention in: +# https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/ +worker_annotations = + + +[kubernetes_node_selectors] +# The Key-value pairs to be given to worker pods. +# The worker pods will be scheduled to the nodes of the specified key-value pairs. +# Should be supplied in the format: key = value + +[kubernetes_environment_variables] +# The scheduler sets the following environment variables into your workers. You may define as +# many environment variables as needed and the kubernetes launcher will set them in the launched workers. +# Environment variables in this section are defined as follows +# = +# +# For example if you wanted to set an environment variable with value `prod` and key +# `ENVIRONMENT` you would follow the following format: +# ENVIRONMENT = prod +# +# Additionally you may override worker airflow settings with the AIRFLOW__
__ +# formatting as supported by airflow normally. + +[kubernetes_secrets] +# The scheduler mounts the following secrets into your workers as they are launched by the +# scheduler. You may define as many secrets as needed and the kubernetes launcher will parse the +# defined secrets and mount them as secret environment variables in the launched workers. +# Secrets in this section are defined as follows +# = = +# +# For example if you wanted to mount a kubernetes secret key named `postgres_password` from the +# kubernetes secret object `airflow-secret` as the environment variable `POSTGRES_PASSWORD` into +# your workers you would follow the following format: +# POSTGRES_PASSWORD = airflow-secret=postgres_credentials +# +# Additionally you may override worker airflow settings with the AIRFLOW__
__ +# formatting as supported by airflow normally. + +[kubernetes_labels] +# The Key-value pairs to be given to worker pods. +# The worker pods will be given these static labels, as well as some additional dynamic labels +# to identify the task. +# Should be supplied in the format: key = value diff --git a/airflow/data/logs/.gitkeep b/airflow/data/logs/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/airflow/data/plugins/.gitkeep b/airflow/data/plugins/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/airflow/docker-compose-master.yml b/airflow/deprecated/docker-compose-master.yml similarity index 100% rename from airflow/docker-compose-master.yml rename to airflow/deprecated/docker-compose-master.yml diff --git a/airflow/docker-compose-worker.yml b/airflow/deprecated/docker-compose-worker.yml similarity index 100% rename from airflow/docker-compose-worker.yml rename to airflow/deprecated/docker-compose-worker.yml diff --git a/airflow/docker-stack.yaml b/airflow/docker-stack.yaml new file mode 100644 index 0000000..4e080f3 --- /dev/null +++ b/airflow/docker-stack.yaml @@ -0,0 +1,110 @@ +version: "3.7" + +services: + + redis: + image: redis:alpine + command: --save 900 1 + ports: + - "6379:6379" + volumes: + - /data/redis:/data + deploy: + replicas: 1 + placement: + constraints: + - node.role == manager + restart_policy: + condition: on-failure + + postgres: + image: postgres:alpine + ports: + - "5432:5432" + volumes: + - /data/postgres:/var/lib/postgresql/data + environment: + - POSTGRES_USER=airflow + - POSTGRES_PASSWORD=airflow + - POSTGRES_DB=airflow + deploy: + replicas: 1 + placement: + constraints: + - node.role == manager + restart_policy: + condition: on-failure + + webserver: + image: vimagick/airflow + command: webserver + ports: + - "8080:8080" + volumes: + - airflow_data:/opt/airflow + deploy: + replicas: 1 + placement: + constraints: + - node.role == manager + restart_policy: + condition: on-failure + depends_on: + - postgres + - redis + + scheduler: + image: vimagick/airflow + command: scheduler + volumes: + - airflow_data:/opt/airflow + deploy: + replicas: 1 + placement: + constraints: + - node.role == manager + restart_policy: + condition: on-failure + depends_on: + - webserver + + flower: + image: vimagick/airflow + command: flower + ports: + - "5555:5555" + volumes: + - airflow_data:/opt/airflow + deploy: + replicas: 1 + placement: + constraints: + - node.role == manager + restart_policy: + condition: on-failure + depends_on: + - webserver + + worker: + image: vimagick/airflow + command: worker + volumes: + - airflow_data:/opt/airflow + deploy: + replicas: 0 + placement: + constraints: + - node.role == worker + restart_policy: + condition: on-failure + depends_on: + - webserver + +volumes: + + airflow_data: + driver: local + driver_opts: + type: nfs + o: "addr=10.0.0.1,nolock,soft,rw" + device: ":/export/airflow"