Suppose several supermarkets deliver daily data at different times. Scheduling a DAG for the latest possible delivery avoids missing files but leaves earlier arrivals waiting. A sensor lets the DAG start sooner and holds each dependent task until its data is available. section 7.1
A sensor is an operator that repeatedly checks a condition. Each check is a poke: a true result succeeds and permits downstream tasks to run; a false result leads to another check after the poke interval. The default interval is 60 seconds, and the sensor fails if its timeout is reached. section 7.1 section 7.1.2
Wait for a delivered file
Use `FileSensor(task_id="wait_for_supermarket_1", filepath="/data/supermarket1/data.csv")` before the task that processes that file. It checks whether the path exists; task logs show each poke. Set `poke_interval` to change the checking frequency, but avoid an interval so short that checks overload the target system. section 7.1
Plan for missing data as well as successful arrivals:
- Set a suitable `timeout`. Its default is seven days; with daily DAG runs, missing deliveries can leave sensors from successive runs active and consume limited task slots. section 7.1.2
- Consider `mode="reschedule"` when waiting would otherwise occupy a slot between pokes. Deferrable operators offer another way to free worker slots while waiting. section 7.1.2 section 9.7
For a condition without a suitable built-in sensor, subclass `BaseSensorOperator` and implement `poke(context)` to return a Boolean. A custom ratings sensor, for example, uses a hook to check whether ratings exist for a templated date range before a fetch task runs. `PythonSensor` is another option for supplying Boolean-checking logic. section 9.4 section 7.4
Listing 9.29 Implementing the poke method
class MovielensRatingsSensor(BaseSensorOperator):
def poke(self, context:Context):
hook = MovielensHook(self._conn_id)
try:
next( #1
hook.get_ratings(
start_date=self._start_date,
end_date=self._end_date,
batch_size=1
)
)
self.log.info(
f"Found ratings for {self._start_date} to {self._end_date}"
)
return True #2
except StopIteration: #3
self.log.info(
f"Didn't find any ratings for {self._start_date} "
f"to {self._end_date}, waiting..."
)
return False #3
finally:
hook.close() #4
See how a custom `poke` method checks for one ratings record, returns true or false, and closes its hook. section 9.4
