**Description**
The Airflow logs generated by the Livy operator currently only state the status of the submitted batch. To view the logs from the job itself, one must go separately to the session logs. I think that Airflow should have the option (possibly on by default) that retrieves the session logs after the batch reaches a terminal state if a `polling_interval` has been set.

**Use case / motivation**
When debugging a task submitted via Livy, the session logs are the first place to check. For most other tasks, including SparkSubmitOperator, viewing the first-check logs can be done in the Airflow UI, but for Livy you must go externally or write a separate task to retrieve them. 

**Are you willing to submit a PR?**

I don't yet have a good sense of how challenging this will be to set up and test. I can try but if anyone else wants to go for it, don't let my attempt stop you.

**Related Issues**

None I could find

airflow/providers/apache/livy/operators/livy.py
:param driver_cores: number of cores to use for the driver process.
    :type driver_cores: str, int
    :param executor_memory: amount of memory to use per executor process.
    :type executor_memory: str
    :param executor_cores: number of cores to use for each executor.
    :type executor_cores: str, int
    :param num_executors: number of executors to launch for this session.
    :type num_executors: str, int
    :param archives: archives to be used in this session.
    :type archives: list
    :param queue: name of the YARN queue to which the application is submitted.
    :type queue: str
    :param name: name of this session.
    :type name: str
    :param conf: Spark configuration properties.
    :type conf: dict
    :param proxy_user: user to impersonate when running the job.
    :type proxy_user: str
    :param livy_conn_id: reference to a pre-defined Livy Connection.
    :type livy_conn_id: str
    :param polling_interval: time in seconds between polling for job completion. Don't poll for values >=0
    :type polling_interval: int
    :param extra_options: A dictionary of options, where key is string and value
        depends on the option that's being modified.
    :type extra_options: Dict[str, Any]
    :param extra_headers: A dictionary of headers passed to the HTTP request to livy.
    :type extra_headers: Dict[str, Any]
    """
    template_fields = ('spark_params',)
    def __init__(