Disable attaching jobs in dataproc using Airflow _DataProcJob hook

user_2336124 0
Viewed 623

im running jobs in GCP dataproc via airflow using gcp_dataproc_hook. before each job execution the hook using check if the job can be attached to a previously executed job.

when a job is attached dataprock dosn't execute the job unless i delete the previous (attached) job

due to this flow i loss a lot of meta and logs.

is there any way to disable the attachment?

this is the code in the hook that creates the attachment:

        # There is a small set of states that we will accept as sufficient
        # for attaching the new task instance to the old Dataproc job.  We
        # generally err on the side of _not_ attaching, unless the prior
        # job is in a known-good state. For example, we don't attach to an
        # ERRORed job because we want Airflow to be able to retry the job.
        # The full set of possible states is here:
        # https://cloud.google.com/dataproc/docs/reference/rest/v1beta2/projects.regions.jobs#State
        recoverable_states = frozenset([
            'PENDING',
            'SETUP_DONE',
            'RUNNING',
            'DONE',
        ])

        found_match = False
        for job_on_cluster in jobs_on_cluster:
            job_on_cluster_id = job_on_cluster['reference']['jobId']
            job_on_cluster_task_id = job_on_cluster_id[:-UUID_LENGTH]
            if task_id_to_submit == job_on_cluster_task_id:

                self.job = job_on_cluster
                self.job_id = self.job['reference']['jobId']
                found_match = True

                # We can stop looking once we find a matching job in a recoverable state.
                if self.job['status']['state'] in recoverable_states:
                    break

        if found_match and self.job['status']['state'] in recoverable_states:
            message = """
    Reattaching to previously-started DataProc job %s (in state %s).
    If this is not the desired behavior (ie if you would like to re-run this job),
    please delete the previous instance of the job by running:

    gcloud --project %s dataproc jobs delete %s --region %s
"""
airflow google-cloud-platform python-3.x
4 Answers

This code has been rearranged in the master branch and I can't find if this feature is still there.

When you submit a job, the DataProc operators add a random UUID to the job ID. I cannot see how code could submit the same job ID with an old UUID. But apparently it does. I have solved the problem by making a copy of the Hook class and commenting out 'DONE' from the list of "recoverable states".

Cheers.

user_14948210

I think, there is no option to set is off and I also had the same issue when I migrated DAGs to Airflow 1.10.11. While creating multiple jobs using DataProc hook in the same airflow task, Here, I managed to solve by setting a unique job name each time which always made the check if task_id_to_submit == job_on_cluster_task_id: False creating new jobs.

    def build_hive_job(self, job_type, **kwargs):
    job = self.hook.create_job_template(task_id="HiveJob",
                                        cluster_name=self.cluster,
                                        job_type="hiveJob",
                                        properties=None)
    if job_type == "hql":
        job.set_job_name(f"HiveTablesCreationJob-{str(uuid.uuid4())}")
        job.add_query_uri(kwargs['hql'])
        job.add_variables(kwargs['variables'])
    elif job_type == "query":
        job.set_job_name(f"HiveDatabaseCreationJob-{str(uuid.uuid4())}")
        job.add_query(kwargs['queries'])
    else:
        raise ValueError("Job type {} is not valid".format(job_type))

    return job.build()
user_38444900

I solved this issue by using DataprocSubmitJobOperator google docs: submit-a-job-to-a-cluster

from airflow.providers.google.cloud.operators.dataproc import DataprocSubmitJobOperator    
        ...................................
    with DAG('your_dag.py',
                 schedule_interval = '0 * * * *', 
                 default_args=DEFAULT_DAG_ARGS
                ) as dag:
    
                PYSPARK_JOB = {
                "reference": {"project_id": project},
                "placement": {"cluster_name": cluster},
                "pyspark_job": {"main_python_file_uri": 'gs://your_bucket/your_app.py'},
                }
                
                pyspark_task = DataprocSubmitJobOperator(
                            task_id="pyspark_task",
                            job=PYSPARK_JOB,
                            location='your_region',
                            project_id= project,
                            dag = dag)
user_148773660

I recently came up with a solution to this problem about the same DataProc JOB ID being generated during the first iteration in a 'DONE' state which leads to the second iteration to fail.

Below is what I did:

bq_to_gcs_spark_job= DataProcPySparkOperator(
task_id='bq_to_gcs_spark_task',
main='<your pyspark code GCS location>',
job_name='bq_to_gcs_spark_job',
cluster_name='<your cluster name>',
gcp_conn_id='<your conn ID>',
region='us-central1',
labels= {keyy: valuee},
dag=dag)

Notice the labels parameter! Set it to a custom key value pair of your choice. This is more of a tag that is visible on the DataProc cluster's job section.

Next, run a BashOperator code on Airflow which looks a bit like this:

cleanup_job = BashOperator(
task_id='cleanup_task',
bash_command= """ printf 'Yes' | gcloud dataproc jobs delete "$(gcloud dataproc jobs list --region=us-central1 --filter='labels.<your key>= <your value>' | awk 'NR==2{print $1}')" --region us-central1 """ ,
dag=dag)

Let me break down the intricate looking shell command.

  1. I am first listing the jobs that have the tag that we defined
  2. Passing the output to the awk command to fetch only the JOB ID since it returns a 3 column table. The awk command fetches the first column ($1) and the first row's value (NR==2) which is basically our JOB ID
  3. Then passed along to the CLI command to delete a DataProc job
  4. A prompt for confirmation (Yes/N) is popped up and during runtime we want it to gracefully continue hence the printf statement.

NOTE:

  1. Kindly keep the key-value pair unique to your own code/use-case to ensure only your job is deleted from DataProc.
  2. This whole will not be required if the Airflow pipeline is creating and then destroying a temporary DataProc cluster
user_46900250
Related