Review contents of the component towards providing an up-to-date userguide and write benchmarking code for generating performance reports (JMH).

BeamSQL has a long list of of aggregation/aggregation analytics functionalities to support.

To begin with, you will need to support this syntax:

analytic_function_name ( [ argument_list ] )
OVER (
[ PARTITION BY partition_expression_list ]
[ ORDER BY expression [

] [, ...] ]
[ window_frame_clause ]
)

This will requires touch core components of BeamSQL:
1. SQL parser to support the syntax above.
2. SQL core to implement physical relational operator.
3. Distributed algorithms to implement a list of functions in a distributed manner.
4. Build benchmarks to measure performance of your implementation

BeamSQL has a long list of of aggregation/aggregation analytics to support.

tests/executors/test_celery_executor.py::TestCeleryExecutor::test_celery_integration_0_amqp_guest_guest_rabbitmq_5672

Log attached.

There is GcsToGDriveOperator but there isn't the equivalent in the other direction

Login page of the UI has autocomplete for username field. This should be disabled for security

1. accept argument exec_date_from, exec_date_to to filter execution_dates returned, ie show dag runs with exec_date between 20190901 and 20190930
2. separate to that in the output print the start_date and end_date of each dagrun (ie execdate for 20190907 had start_date 2019090804:23 and end_date 2019090804:38
3. dag_id arg should be optional

Cassandra consume 1.173GiB of memory. Travis does not have very efficient machines, so we should limit system/integration tests of components that do not require much attention, e.g. they are not changed often. Cassandra is a good candidate for this. This will allow the machine power to be used for more needed work.

            CONTAINER ID        NAME                                  CPU %               MEM USAGE / LIMIT     MEM %               NET
            I/O             BLOCK I/O           PIDS
            8aa37ca50f7c        ci_airflow-testing_run_1f3aeb6d1052   0.00%               5.715MiB / 3.855GiB   0.14%               1.14kB
            / 0B         2.36MB / 0B         2
            f2b3be15558f        ci_cassandra_1                        0.69%               1.173GiB / 3.855GiB   30.42%              2.39kB
            / 0B         75.3MB / 9.95MB     50
            ef1de3981ca6        ci_krb5-kdc-server_1                  0.02%               12.15MiB / 3.855GiB   0.31%               2.46kB
            / 0B         18.9MB / 184kB      4
            be808233eb91        ci_mongo_1                            0.31%               36.71MiB / 3.855GiB   0.93%               2.39kB
            / 0B         43.2MB / 19.1MB     24
            667e047be097        ci_rabbitmq_1                         0.77%               69.95MiB / 3.855GiB   1.77%               2.39kB
            / 0B         43.2MB / 508kB      92
            2453dd6e7cca        ci_postgres_1                         0.00%               7.547MiB / 3.855GiB   0.19%               1.05MB
            / 889kB      35.4MB / 145MB      6
            78050c5c61cc        ci_redis_1                            0.29%               1.695MiB / 3.855GiB   0.04%               2.46kB
            / 0B         6.94MB / 0B         4
            c117eb0a0d43        ci_mysql_1                            0.13%               452MiB / 3.855GiB     11.45%              2.21kB
            / 0B         33.9MB / 548MB      21
            131427b19282        ci_openldap_1                         0.00%               45.68MiB / 3.855GiB   1.16%               2.64kB
            / 0B         32.8MB / 16.1MB     4
            8c2549c010b1        ci_docker_1                           0.59%               22.06MiB / 3.855GiB   0.56%               2.39kB
            / 0B         95.9MB / 291kB      30
            

Airflow has SalesForceHook but it doesn't have a distinct connection.

In order to create a Connection one must expose it's secret token as text :

https://stackoverflow.com/questions/53510980/salesforce-connection-using-apache-airflow-ui

Also it's not very intuitive that the Conn Type should remain blank.

It would be easier and also user friendly if there will be salesforce connection in the UI which has a security_token field that is encrypted.

dag level logs are written under separate date folders. This is great because the old dates are not 'modified/accessed' so they can be easily purged by utilities like tmpwatch

This JIRA is about making other logs (such as dag_processor_manager/webserver/scheduler.etc) go under separate date folders to allow easy purging. the log from redirecting 'airflow scheduler' to stdout grows over 100mb a day in my env

https://www.oktopost.com/

key-value pairs of conf can be passed into trigger_dag command
ie
--conf '

'

clear command needs this feature too

ie in case exec_date is important and there was a failure halfway in the 1st dagrun due to bad conf being sent on trigger_dag command and want to run the same execdate but with new conf on 2nd dagrun

alternative solution would be a new delete_dag_run cli command so never need to 'clear' but can do a 2nd DagRun for same exec date

Multiline attributes, rendered templates, or Xcom variables are not well supported in the admin GUI at present. Any values are treated as native HTML text() blocks and as such all formatting is lost. When passing structured data such as YAML in these variables, it makes a real mess of them.

Ideally, these values should keep their line-breaks and indentation.

This should only require having these code blocks wrapped in a <pre> block or setting `white-space: pre` on the class for the block.

I am getting an error when passing in a SQL file with multiple SQL statements to snowflake operator

            snowflake.connector.errors.ProgrammingError: 000006 (0A000): 01908236-01a3-b2c4-0000-f36100052686: Multiple SQL statements
            in a single API call are not supported; use one API call per statement instead.
            

It only fails if you pass a file with multiple statements. A file with just one statement or list of statements to the operator works fine.

After looking at the current snowflake operator implementation it seems like a list of SQL statements work because it executes one statement at a time. Whereas multiple statements in a SQL file fails because all of them are read as one continuous string.

How can we fix this:

There is an API call in Snowflake python connector that supports multiple SQL statements.

https://docs.snowflake.net/manuals/user-guide/python-connector-api.html#execute_string

This can be fixed by overriding the run function in Snowflake Hook to support multiple sql statements in a file.

The SageMakerEndpointOperator currently taken an argument "operati on" with value "create"/"update" which determines whether to create or update a SageMaker endpoint. However this doesn't work in the following situation:

• DAG run #1 create the endpoint (have to provide operation="create" here)
• Following DAG runs will update the endpoint created by DAG run #1 (would have to edit DAG and set operation="update" here)

Which should be a very valid use case IMO.

The SageMakerEndpointOperator should check itself if an endpoint with name X already exists and overwrite it (configurable desired by the user).

Currently, in the main view, the schedule column box is highlighted in red if the max. number of DAG runs is achieved. In this case no more DAG runs can be started until a DAG run completes.

I think it should also display in red when the dag_concurrency (i.e. max concurrent tasks) is achieved. In this case also, no more tasks can be started until a task completes. However there is currently nothing in the UI showing that (currently running 1.10.5).

The fact that I can specify a minimum modified time to filter objects on in GoogleCloudStorageToGoogleCloudStorageOperator but not a maximum seems rather arbitrary. Especially considering the typical usage scenario of running a copy on a schedule, I would like to be able to find objects created within a particular schedule interval for my execution, and not just copy all of the latest objects.

The restriction in GoogleCloudStorageToGoogleCloudStorageOperator that I am only allowed to specify a single object to list is rather arbitrary. If I specify a wildcard, all it does is split at the wildcard and use that to get a prefix and delimiter. Why not just let me do this search myself and return a list of objects?

I am proposing a new config flag. It will enforce a generic override in all airflow logging to suppress printing any lines containing case-insensitive match on any of: password|secret|credential|token

If you do a

            grep -iE 'password|secret|credential|token' -R <airflow_logs_folder>

you may be surprised with what you find :O

ideally could replace only the sensitive value but there are various formats like:  

            key=value, key'=value, key value, key"=value, key = value, key"="value, key:value

..etc

AWSBatchOperator does not currently support AWS Batch Job parameters.

When creating an AWS Batch Job Definition and when submitting a job to AWS Batch, it's possible to define and supply job parameters. Most of our AWS Batch jobs take parameters but we are not able to pass them using the AWSBatchOperator.

In order to support batch job parameters, a new argument to _init_(self) could be added called job_parameters, saved to an instance variable and supplied to self.client.submit_job() in the execute() method:

            self.client.submit_job(
            jobName=self.job_name,
            jobQueue=self.job_queue,
            jobDefinition=self.job_definition,
            containerOverrides=self.overrides,
            parameters=self.job_parameters)
            

See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/batch.html#Batch.Client.submit_job

I'm not sure how this is doable but one of the key components that is missing in airflow is the ability to notify about detecting anomalies something like graphana https://grafana.com/

It would be great if airflow can add support for such tools

I'm talking here about airflow itself. For example: if DAG run normally takes 5 minutes but now for any reason it's running over 30 minutes than we want an alert to be sent with graph that shows that anomaly.

Per discussion in https://github.com/apache/airflow/pull/5664 we might use dmypy for local development speedups.

`GoogleCloudStorageToBigQueryOperator` should be able to have its `source_objects` dynamically determined by the results of a previous workflow. This is hard to do with it expecting a list, as any template expansion will render as a string. This could be implemented either as a check for whether `source_objects` is a string, and trying to parse it as a list if it is, or a separate argument for a string encoded as a list.

My particular use case for this is as follows:

1. A daily DAG scans a GCS bucket for all objects created in the last day and loads them into BigQuery.
2. To find these objects, a `PythonOperator` scans the bucket and returns a list of object names.
3. A `GoogleCloudStorageToBigQueryOperator` is used to load these objects into BigQuery.

The operator should be able to have its list of objects provided by XCom, but there is no functionality to do this, and trying to do a template expansion along the lines of `source_objects='{{ task_instance.xcom_pull(key="KEY") }}'` doesn't work because this is rendered as a string, which `GoogleCloudStorageToBigQueryOperator` will try to treat as a list, with each character being a single item.

Make AWS Operators Pylint compatible.

Hi

This project lacks integration with the Google Search Ads 360 service. I would be happy if Airflow had proper operators and hooks that integrate with this service.

Product Documentation: https://developers.google.com/search-ads/
API Documentation: https://developers.google.com/resources/api-libraries/documentation/dfareporting/v3.3/python/latest/

Lots of love

Hi

This project lacks integration with the Cloud AutoML Tables service. I would be happy if Airflow had proper operators and hooks that integrate with this service.

Product Documentation: https://cloud.google.com/automl-tables/docs/
API Documentation: https://googleapis.github.io/google-cloud-python/latest/automl/index.html

Love

Hi

This project lacks integration with the Cloud AutoML NL Sentiment service. I would be happy if Airflow had proper operators and hooks that integrate with this service.

Product Documentation: https://cloud.google.com/natural-language/automl/sentiment/docs/
API Documentation: https://googleapis.github.io/google-cloud-python/latest/automl/index.html

Lots of love

Hi

This project lacks integration with the Cloud AutoML NL Entity Extraction service. I would be happy if Airflow had proper operators and hooks that integrate with this service.

Product Documentation: https://cloud.google.com/natural-language/automl/entity-analysis/docs/
API Documentation: https://googleapis.github.io/google-cloud-python/latest/automl/index.html

Love

Hi

This project lacks integration with the Cloud AutoML NL Classification service. I would be happy if Airflow had proper operators and hooks that integrate with this service.

Product Documentation: https://cloud.google.com/natural-language/automl/docs/
API Documentation: https://googleapis.github.io/google-cloud-python/latest/automl/index.html

Love

Allow to load compressed file in the load_file function.

We have similar logic in GoogleCloudStorageHook

Add hook to interact with FacebookAds

We have a use case where we are uploading files to an S3 bucket in a different AWS account to the one Airflow is running in.  AWS S3 supports this situation using the pre canned ACL policy, specifically bucket-owner-full-control. 

However, the current implementations of the S3Hook.load_*() and S3Hook.copy_object() methods do not allow us to supply any ACL policy for the file being uploaded/copied to S3.  

It would be good to add another optional parameter to the S3Hook methods called acl_policy which would then be passed into the boto3 client method calls like so 

            # load_file
            ...
            if encrypt: 
            extra_args['ServerSideEncryption'] = "AES256"
            if acl_policy:
            extra_args['ACL'] = acl_policy
            
            client.upload_file(filename, bucket_name, key, ExtraArgs=extra_args)

            # load_bytes
            ...
            if encrypt: 
            extra_args['ServerSideEncryption'] = "AES256"
            if acl_policy:
            extra_args['ACL'] = acl_policy
            
            client.upload_file(filename, bucket_name, key, ExtraArgs=extra_args)

            # copy_object
            self.get_conn().copy_object(Bucket=dest_bucket_name,
            Key=dest_bucket_key,
            CopySource=CopySource, 
            ACL=acl_policy)
            

The "Browse Logs" UI currently allows filtering by "DAG ID", "Task ID", "Execution Date", and "Extra".

For consistency and flexibility, it would be good to allow filtering by any of the available columns, specifically "Datetime", "Event", "Execution Date", and "Owner". 

Currently It seems that K8 Executor expects the dags_volume_claim or git_repo to be always defined through airflow.cfg. Otherwise it does not come up.
Though there is support for "emptyDir" volume in worker_configuration.py but kubernetes_executor fails in _validate function if these configs are not defined.
Our dag files are stored in some remote location which can be synced to worker pod through init/side-car container. We are exploring if it makes sense to allow K8 executor to come up for cases where dags_volume_claim are git_repo are not defined. In such cases worker pod would look for the dags in emptyDir and worker_airflow_dags path (like it does for git-sync). Dag files can be made available in worker_airflow_dags path through init/side-car container.

The cc1e65623dc7_add_max_tries_column_to_task_instance migration creates a DagBag for the corresponding DAG for every single task instance. This is very redundant and not necessary.

Hence, there are discussions on Slack like these:

murquizo   [Jan 17th at 1:33 AM]
            Why does the airflow upgradedb command loop through all of the dags?
            
            ....
            
            murquizo   [14 days ago]
            NICE, @BasPH! that is exactly the migration that I was referring to.  We have about 600k task instances and have a several
            python files that generate multiple DAGs, so looping through all of the task_instances to update max_tries was too slow. 
            It took 3 hours and didnt even complete! i pulled the plug and manually executed the migration.   Thanks for your response.
            

An easy to accomplish improvement is to parse a DAG only once and after that set the task instance try_number. I created a branch for it (https://github.com/BasPH/incubator-airflow/tree/bash-optimise-db-upgrade), currently running tests and will make PR when done.

The env_vars attribute of the KubernetesPodOperator allows to pass environment variables as string but it doesn't allows to pass a value from a configmap or a secret.

I'd like to be able to do

            modeling = KubernetesPodOperator(
            ...
            env_vars={
            'MY_ENV_VAR': {
            'valueFrom': {
            'secretKeyRef': {
            'name': 'an-already-existing-secret',
            'key': 'key',
            }
            }
            },
            ...
            )
            

Right now if I do that, Airflow generates the following config

            - name: MY_ENV_VAR
            value:
            valueFrom:
            configMapKeyRef:
            name: an-already-existing-secret
            key: key
            

instead of 

            - name: MY_ENV_VAR
            valueFrom:
            configMapKeyRef:
            name: an-already-existing-secret
            key: key
            

The extract_env_and_secrets method of the KubernetesRequestFactory could check if the value is a dictionary and use it directly.

Is there a way to provide env variables while launching K8 pod through K8 executor. we need to pass some env variable which are referred inside our Airflow Operator. so can we provide custom env variable to docker run command while launching task pod. Currently it seems that it supports predefined env variable.

worker_configuration.py

def get_environment(self): """Defines any necessary environment variables for the pod executor""" env = { 'AIRFLOWCOREDAGS_FOLDER': '/tmp/dags', 'AIRFLOWCORE_EXECUTOR': 'LocalExecutor' } if self.kube_config.airflow_configmap: env['AIRFLOW__CORE__AIRFLOW_HOME'] = self.worker_airflow_home return env

Possible solution

At the moment there is not a way to configure environmental variables on a per-task basis, but it shouldn't be too hard to add that functionality. Extra config options can be passed through the `executor_config` on any operator:

https://github.com/apache/incubator-airflow/blob/master/airflow/models.py#L2423-L2437

Which are eventually used here to construct the kubernetes pod for the task:

https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/kubernetes/worker_configuration.py#L186

The current HdfsHook relies on the snakebite library, which is unfortunately not compatible with Python 3. To add Python 3 support for the HdfsHook requires switching to a different library for interacting with HDFS. The hdfs3 library is an attractive alternative, as it supports Python 3 and seems to be stable and relatively well supported.

Update: hdfs3 doesn't get any updates anymore. The best library right now seems to be pyarrow: https://arrow.apache.org/docs/python/filesystems.html
Therefore I would like to upgrade to pyarrow instead of hdfs3.

In the UI there is a search box.

If you search DAG name you will see the result for the search as you type.

Please allow support of wild-cards. Mainly for : *

So if I have a Dag called :abcd and I'm searching for ab* I will see it in the list.

This is very helpful for systems with 100+ dags.

When using HttpSensor, which underlying would use HttpHook to perform the request. If the target service is down, which refused the connection, the task would fail immediately.

would be great if this behaviour is configurable, so the sensor would keep sensoring until the service is up again.

traceback of the error:
[2017-04-29 02:00:31,248]

INFO - Subtask: requests.exceptions.ConnectionError: HTTPConnectionPool(host='xxxx', port=123): Max retries exceeded with url: /xxxx (Caused by NewConnectionError('<requests.packages.urllib3.connection.HTTPConnection object at 0x7f94b64b44e0>: Failed to establish a new connection: [Errno 111] Connection refused',))

Please provide a /health endpoint of each of the following component:

• webservice (to avoid pinging the / root endpoint)
• worker
• scheduler

This would ease integration in Mesos/Marathon framework.

If you agree, I volunteer to add this change.

Add ability to specify custom parameters to docker cli. E.g. "--volume-driver=""" or --net="bridge" or any other