Confusing usage of AthenaSensor?

[saswatarautray]

As per the various blogs on AthenaSensor, the usage looks like:

  task1 = AthenaOperator (helps querying on aws athena.)
  task2 = AthenaSensor (waits until task1 is completed.)
  task3 = PythonOperator
  task1 >> task2 >> task3

However, in case of sequential executor, only 1 task can run at a time. We can specify the order of execution of tasks using Bitwise (<</>>) operators, which means task3 cannot run before task1 completes.

So, is the need for AthenaSensor redundant? What's its use?

[potiuk]

It's not confusing at all. What actually is confusing is to see you use Sequential Executor.

The SequentialExecutor is the default executor when you first install airflow. It is the only executor that can be used with sqlite since sqlite doesn’t support multiple connections. This executor will only run one task instance at a time. For production use case, please use other executors.

Could you please explaine @saswatarautray what made you use Sequential Executor despite warning in the documentation and big fat warnings that you see when you run airlfow webserver. I am sincerely confused. Could you please explain what made you make such an odd choice?

[alexkruc]

First of all, you shouldn't use SequentialExecutor in production usage, but since here task2 is dependent on the completion of task1 it doesn't really matter as tasks will still run one after another..

I guess the reason that you might consider using AthenaSensor immediately after AthenaOperator, is if you specified a max_tries parameter with a value that is low. According to the operator's code, it will try to poll the job state for the time specified in max_tries, but the operator does not actually verify the query succession at all times. It means that even after task1 finishing in Airflow, it might be still running in Athena, and to act according to the data, you need to verify its completion using a sensor.

Also, don't forget that sensors could be also independent.. Another DAG's task1 can be only the sensor, that waits for some specific resource from Athena, and acts by it.. So a sensor task is needed if you are depending on someone else AthenaOperator.

Finally, it's hard to say why these blogs mention this task's skeleton without seeing the actual blog posts.. But maybe the intention was purely educational, to show you many ways that you can use Athena from Airflow.

[atiannicelli]

I guess the reason that you might consider using AthenaSensor immediately after AthenaOperator, is if you specified a max_tries parameter with a value that is low. According to the operator's code, it will try to poll the job state for the time specified in max_tries, but the operator does not actually verify the query succession at all times. It means that even after task1 finishing in Airflow, it might be still running in Athena, and to act according to the data, you need to verify its completion using a sensor.

I don't understand this. Are you saying that Task1 can complete with success and the task is not complete? I have been experimenting with the Athena Operator and it seems that when the AthenaOperator completes that the data that was produced by that task is available and the next operator can just wait on the athenOperator without having to add in a sensor there. I totally get the use of the sensor when you are waiting on an operator from a different pipeline/flow, but within the same flow it seems all I need to do is make each take depend on the last and it works... What am I missing?

[alexkruc]

Well, my comment was almost two years ago, and a lot of stuff and features were added since then :) If I recall properly, 2 years ago, AthenaOperator just fired a request to AWS to run a query or something, but didn't actually validate it's final state.. This is why people used the sensor immediately after the operator to verify that their data is ready.

If you have a different experience now, that's awesome! :) I don't personally use Athena, so I cannot verify it myself, but it's good that in your case, once the operator finishes, the data is already available.

[kien-truong]

Per the current implementation, AthenaSensor is a bit useless if you're already using AthenaOperator, because AthenaOperator has already waited for query completion and will raise exception if the query did not complete.

airflow/providers/amazon/src/airflow/providers/amazon/aws/operators/athena.py

Lines 165 to 175 in 5d87bdd

	if query_status in AthenaHook.FAILURE_STATES:
	error_message = self.hook.get_state_change_reason(self.query_execution_id)
	raise AirflowException(
	f"Final state of Athena job is {query_status}, query_execution_id is "
	f"{self.query_execution_id}. Error: {error_message}"
	)
	elif not query_status or query_status in AthenaHook.INTERMEDIATE_STATES:
	raise AirflowException(
	f"Final state of Athena job is {query_status}. Max tries of poll status exceeded, "
	f"query_execution_id is {self.query_execution_id}."
	)