使用 Airflow _DataProcJob 钩子禁用在 dataproc 中附加作业

Question

我使用 gcp_dataproc_hook 通过气流在 GCP 数据过程中运行作业。 在每个作业执行之前，使用检查该作业是否可以附加到先前执行的作业的钩子。

当附加作业时，除非我删除前一个（附加的）作业，否则 dataproc 不会执行该作业

由于这个流程，我丢失了很多元数据和日志。

有没有办法禁用附件？

这是创建附件的钩子中的代码：

        # 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
"""

Answer 1

这段代码已经在master分支中重新排列了，我找不到这个功能是否还在。

当您提交作业时，DataProc 运算符会向作业 ID 添加一个随机 UUID。我看不到代码如何使用旧的 UUID 提交相同的作业 ID。但显然确实如此。我通过制作 Hook 类的副本并从“可恢复状态”列表中注释掉“DONE”来解决了这个问题。

干杯。

Answer 2

我认为，没有选项设置为关闭，当我将 DAG 迁移到 Airflow 1.10.11 时，我也遇到了同样的问题。 在同一个气流任务中使用 DataProc 钩子创建多个作业时，在这里，我设法通过每次设置一个唯一的作业名称来解决，这总是检查 if task_id_to_submit == job_on_cluster_task_id: False 创建新作业。

    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()

Answer 3

我通过使用 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)

Answer 4

我最近想出了一个解决此问题的方法，即在第一次迭代期间生成的相同 DataProc JOB ID 处于“完成”状态，这会导致第二次迭代失败。

以下是我所做的：

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)

注意labels 参数！将其设置为您选择的自定义键值对。这更像是在 DataProc 集群的作业部分可见的标记。

接下来，在 Airflow 上运行 BashOperator 代码，看起来有点像这样：

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)

让我分解一下看起来很复杂的 shell 命令。

1. 我首先列出具有我们定义的标签的作业
2. 将输出传递给awk 命令以仅获取JOB ID，因为它返回一个3 列的表。 awk 命令获取第一列 ($1) 和第一行的值 (NR==2)，这基本上是我们的 JOB ID
3. 然后传递给 CLI 命令以删除 DataProc 作业
4. 会弹出确认提示 (Yes/N)，在运行时我们希望它正常继续，因此使用 printf 语句。

注意：

1. 请为您自己的代码/用例保留唯一的键值对，以确保仅从 DataProc 中删除您的作业。
2. 如果 Airflow 管道正在创建然后销毁临时 DataProc 集群，则不需要整个过程