I'm trying to write my first dag in Airflow using pySpark in Dataproc, but I don't see a job's log message in the AirFlow UI. I see these messages in Dataproc UI. When I use gcloud dataproc submit pyspark jobs commands I see messages in console.
My dag
with DAG(
'dag_1',
default_args={'depends_on_past': False},
schedule_interval=timedelta(days=1),
start_date=days_ago(2),
tags=['example'],
) as dag:
task = DataprocSubmitJobOperator(
task_id='task_1', region=region, project_id=project_id,
job={
"reference": {"project_id": project_id},
"placement": {"cluster_name": cluster_name},
"pyspark_job": {
"main_python_file_uri": pyspark_uri
},
}
)
task
My task
import logging
from pyspark.sql import SparkSession
if __name__ == '__main__':
logging.basicConfig()
logger = logging.getLogger("airflow.task")
logger.info('logger.info')
spark = (SparkSession
.builder
.appName('test')
.getOrCreate())
print('print')
logger.critical('logger.critical')
In Airflow UI I have only this
[2022-04-28, 12:50:02 UTC] {taskinstance.py:1446} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=dag_1
AIRFLOW_CTX_TASK_ID=task_1
AIRFLOW_CTX_EXECUTION_DATE=2022-04-27T00:00:00+00:00
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2022-04-27T00:00:00+00:00
[2022-04-28, 12:50:02 UTC] {dataproc.py:1847} INFO - Submitting job
[2022-04-28, 12:50:02 UTC] {credentials_provider.py:312} INFO - Getting connection using `google.auth.default()` since no key file is defined for hook.
[2022-04-28, 12:50:02 UTC] {dataproc.py:1859} INFO - Job 02d046aa-6e45-4889-a549-b2dcd44e9603 submitted successfully.
[2022-04-28, 12:50:02 UTC] {dataproc.py:1864} INFO - Waiting for job 02d046aa-6e45-4889-a549-b2dcd44e9603 to complete
[2022-04-28, 12:50:44 UTC] {dataproc.py:1868} INFO - Job 02d046aa-6e45-4889-a549-b2dcd44e9603 completed successfully.
I have Python 3.9 and Airflow 2.2.5