[D] Best way of processing historical data without catchup [airflow]

["akomisarek (via GitHub)" <gi...@apache.org>]

GitHub user akomisarek created a discussion: Best way of processing historical data without catchup

I am looking for a way to process a data gap, without using catchup mechanism, i.e. my DAG's start_date is 2020 and schedule is daily, but I don't want to create 1.5k DAG runs for it. Ideally, I would like one execution, which simply receives data_start_interval from start_date to properly calculated data_interval_end. I.e. my initial DAG Run would receive data_interval_start (2020-01-01) and data_interval_end ( 2025-01-26) it would allow the clean code from the processing perspective to be written which simply analyses the data from specified time-window. 

Is it possible with current Timetables? From my initial testing it seems no, is the recommendation here to build customer timetable? How others are handling such scenarios? Thanks! :) 

GitHub link: https://github.com/apache/airflow/discussions/46141

----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org

Re: [D] Best way of processing historical data without catchup [airflow]

["akomisarek (via GitHub)" <gi...@apache.org>]

GitHub user akomisarek edited a comment on the discussion: Best way of processing historical data without catchup

Answerwing to myself - we think about implementing custom timetable this seems to work well enough. 
POC code:
```
    def next_dagrun_info(self, last_automated_data_interval: DataInterval, restriction: TimeRestriction) -> DagRunInfo:
        if last_automated_data_interval is None:
            if not restriction.catchup:
                next_start = pendulum.now().start_of('day')
                next_end = next_start.add(days=1)
            else:
                next_start = restriction.earliest
                diff_days = (pendulum.now() - next_start).in_days()

                if diff_days >= 365:
                    next_end = next_start.add(years=1)
                else:
                    next_end = next_start.add(days=1)

        else:
            next_start = last_automated_data_interval.end
            diff_days = (pendulum.now() - next_start).in_days()
            if diff_days >= 365:
                next_end = next_start.add(years=1)
            else:
                next_end = next_start.add(days=1)

        return DagRunInfo.interval(start=next_start, end=next_end)
```

GitHub link: https://github.com/apache/airflow/discussions/46141#discussioncomment-12072398

----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org

Re: [D] Best way of processing historical data without catchup [airflow]

["akomisarek (via GitHub)" <gi...@apache.org>]

GitHub user akomisarek added a comment to the discussion: Best way of processing historical data without catchup

Answerwing to myself - we think about implementing custom timetable this seems to work well enough. 

GitHub link: https://github.com/apache/airflow/discussions/46141#discussioncomment-12072398

----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org