**Apache Airflow version**:  1.10.10


**Kubernetes version (if you are using kubernetes)**: 1.14.10

**Environment**:

- **Cloud provider or hardware configuration**: Any, GCP 
- **OS** (e.g. from /etc/os-release): Ubuntu 1.10.10
- **Kernel** (e.g. `uname -a`): Any
- **Install tools**: Any
- **Others**: NA

**What happened**:
DAGs with ExternalTaskMarker don't clean external task after second usage of clean on whole DAG


**What you expected to happen**:

All external task should be cleaned

**How to reproduce it**:
enable serialization `store_serialized_dags = True`

create example DAGs:

```
default_args = {'owner': 'airflow',
                'start_date': datetime(2018, 1, 1)}

def hello_world_py(*args):
    print('Hello World')
    print('This is DAG is dep}')

schedule = '@daily'
dag_id = 'dep_dag'
with DAG(dag_id=dag_id,
         schedule_interval=schedule,
         default_args=default_args) as dag:
    t1 = PythonOperator(task_id='hello_world',
                        python_callable=hello_world_py,)
    dep_1 = ExternalTaskSensor(task_id='child_task1',
                               external_dag_id='hello_world_2',
                               external_task_id='parent_task',
                               mode='reschedule')
    dep_1 >> t1

def create_dag(dag_id, schedule, dag_number, default_args):
    dag = DAG(dag_id, schedule_interval=schedule,
              default_args=default_args)
    with dag:
        t1 = PythonOperator(task_id='hello_world',
                            python_callable=hello_world_py,
                            dag_number=dag_number)
        parent_task = SerializableExternalTaskMarker(task_id='parent_task',
                                         external_dag_id='dep_dag',
                                         external_task_id='child_task1')
        t1 >> parent_task
    return dag

for n in range(1, 4):
    dag_id = 'hello_world_{}'.format(str(n))
    default_args = {'owner': 'airflow',
                    'start_date': datetime(2018, 1, 1)}
    schedule = '@daily'
    dag_number = n
    globals()[dag_id] = create_dag(dag_id, schedule, dag_number, default_args)
```

1. Run both DAGs
2. Wait until first few dagruns where completed
3. Clean first dugrun in DAG with marker
4. Check external dug was cleaned on this date
5. Mark success this date in each DAGs or wait until complete
6. Clean DAG with marker second time on same date
7. ExternalTaskMarker don't work

**Anything else we need to know**:
I think ExternalTaskMarker don't work because of serialization, after serialization each task instance get operator field equal 'SerializedBaseOperator' and markers logic dot' work [here](https://github.com/apache/airflow/blob/master/airflow/models/dag.py#L1072) 

To test ExternalTaskMarker with serialization you can use:

<details>

```
from airflow.sensors.external_task_sensor import ExternalTaskMarker
class FakeName(type):

    def __new__(metacls, name, bases, namespace, **kw):
         name = namespace.get("__name__", name)
         return super().__new__(metacls, name, bases, namespace, **kw)
class SerializableExternalTaskMarker(ExternalTaskMarker, metaclass=FakeName):

    # The _serialized_fields are lazily loaded when get_serialized_fields() method is called
    __serialized_fields = None  # type: Optional[FrozenSet[str]]
    __name__ = "ExternalTaskMarker"

    @classmethod
    def get_serialized_fields(cls):
        """Serialized BigQueryOperator contain exactly these fields."""
        if not cls.__serialized_fields:
            cls.__serialized_fields = frozenset(
                ExternalTaskMarker.get_serialized_fields() | {
                    "recursion_depth", "external_dag_id", "external_taskid", "execution_date"
                }
            )
        return cls.__serialized_fields
```

</details>

@kaxil 
airflow/sensors/external_task_sensor.py
"""
    Waits for a different DAG or a task in a different DAG to complete for a
    specific execution_date
    :param external_dag_id: The dag_id that contains the task you want to
        wait for
    :type external_dag_id: str
    :param external_task_id: The task_id that contains the task you want to
        wait for. If ``None`` (default value) the sensor waits for the DAG
    :type external_task_id: str or None
    :param allowed_states: list of allowed states, default is ``['success']``
    :type allowed_states: list
    :param failed_states: list of failed or dis-allowed states, default is ``None``
    :type failed_states: list
    :param execution_delta: time difference with the previous execution to
        look at, the default is the same execution_date as the current task or DAG.
        For yesterday, use [positive!] datetime.timedelta(days=1). Either
        execution_delta or execution_date_fn can be passed to
        ExternalTaskSensor, but not both.
    :type execution_delta: Optional[datetime.timedelta]
    :param execution_date_fn: function that receives the current execution date
        and returns the desired execution dates to query. Either execution_delta
        or execution_date_fn can be passed to ExternalTaskSensor, but not both.
    :type execution_date_fn: Optional[Callable]
    :param check_existence: Set to `True` to check if the external task exists (when
        external_task_id is not None) or check if the DAG to wait for exists (when
        external_task_id is None), and immediately cease waiting if the external task
        or DAG does not exist (default value: False).
    :type check_existence: bool
    """