[
https://issues.apache.org/jira/browse/AIRFLOW-3354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16689197#comment-16689197
]
Jakub Powierza commented on AIRFLOW-3354:
-----------------------------------------
Here is my reproduction for this issue:
{code:java}
from typing import Mapping, Any
from datetime import datetime
from airflow import DAG
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
default_args = {
'owner': 'gta',
'email': ['my_acco...@gmail.com'],
'retries': 3,
'start_date': datetime(2017, 3, 1),
'depends_on_past': False,
}
dag = DAG('hello_world', default_args=default_args, catchup=False,
schedule_interval='*/1 * * * *')
class MyOperator(BaseOperator):
@apply_defaults
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.start_date = datetime.utcnow()
def execute(self, context: Mapping) -> None:
print('Hello, World!')
hello_world_op = MyOperator(task_id='hello_world_op', dag=dag)
{code}
It seems that setting start date with datetime module inside my operator breaks
scheduler. This change fixes this issue:
{code:java}
from airflow.utils import timezone
...
class MyOperator(BaseOperator):
@apply_defaults
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.start_date = timezone.utcnow()
{code}
However, change in default args is not needed and it works well.
> Scheduler compares offset-naive and offset-aware dates
> ------------------------------------------------------
>
> Key: AIRFLOW-3354
> URL: https://issues.apache.org/jira/browse/AIRFLOW-3354
> Project: Apache Airflow
> Issue Type: Bug
> Affects Versions: 1.10.1
> Reporter: Jakub Powierza
> Priority: Major
>
> New version of Airflow (1.10.1rc1 and 1.10.1rc2) tries to compare
> offset-naive and offset-aware dates in Scheduler. I've tested a simple case
> with schedule set to "*/10 * * * *". I've tried to clean my developer
> instance with `airflow resetdb` and start from scratch but it does not help
> at all. This issue does not occur on stable version 1.10.0.
> My setup: Python 3.6 on Ubuntu 14.04 with Airflow Scheduler based on Celery
> with RabbitMQ backend.
> Exception found in Scheduler logs:
> {code:java}
> 2018-11-15 14:41:23,194:ERROR:airflow.processor:[CT=None] Got an exception!
> Propagating...
> Traceback (most recent call last):
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 389, in helper
> pickle_dags)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 1846, in process_file
> self._process_dags(dagbag, dags, ti_keys_to_schedule)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 1426, in _process_dags
> dag_run = self.create_dag_run(dag)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 909, in create_dag_run
> external_trigger=False
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/models.py",
> line 4270, in create_dagrun
> run.verify_integrity(session=session)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 70, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/models.py",
> line 5215, in verify_integrity
> if task.start_date > self.execution_date and not self.is_backfill:
> TypeError: can't compare offset-naive and offset-aware datetimes
> Process DagFileProcessor40-Process:
> Traceback (most recent call last):
> File "/usr/lib/python3.6/multiprocessing/process.py", line 249, in _bootstrap
> self.run()
> File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
> self._target(*self._args, **self._kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 389, in helper
> pickle_dags)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 1846, in process_file
> self._process_dags(dagbag, dags, ti_keys_to_schedule)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 1426, in _process_dags
> dag_run = self.create_dag_run(dag)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/jobs.py",
> line 909, in create_dag_run
> external_trigger=False
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 74, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/models.py",
> line 4270, in create_dagrun
> run.verify_integrity(session=session)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/utils/db.py",
> line 70, in wrapper
> return func(*args, **kwargs)
> File
> "/home/jpowierz/my_project/venv/lib/python3.6/site-packages/airflow/models.py",
> line 5215, in verify_integrity
> if task.start_date > self.execution_date and not self.is_backfill:
> TypeError: can't compare offset-naive and offset-aware datetimes
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
