Error in passing metadata to DataprocClusterCreateOperator

user_15441104 0
Viewed 147

I am using Airflow version: 2.1.4+composer and I have a DAG where I defined the DataprocClusterCreateOperator like this:

create_dataproc =  dataproc_operator.DataprocClusterCreateOperator(
  task_id='create_dataproc',
  cluster_name='dataproc-cluster-demo-{{ ds_nodash }}',
  num_workers=2,
  region='us-east4',
  zone='us-east4-a',
  subnetwork_uri='projects/example',
  internal_ip_only=True,
  tags=['allow-iap-ssh'],
  init_actions_uris=['gs://goog-dataproc-initialization-actions-us-east4/connectors/connectors.sh'],
  metadata=[('spark-bigquery-connector-url','gs://spark-lib/bigquery/spark-2.4-bigquery-0.23.1-preview.jar')],
  labels=dict(equipo='dm',ambiente='dev',etapa='datapreparation',producto='x',modelo='x'),
  master_machine_type='n1-standard-1',
  worker_machine_type='n1-standard-1',
  image_version='1.5-debian10'
  )

I passed the metadata as a sequence of tuples as I read here, using the dict is not working.

Also, the metadata is not being rendered in the cluster_config when I check the log.

The error in the log saved into the bucket is: "echo 'ERROR: None of connector versions or URLs are specified'"

How can I pass the metadata parameter to this operator and make it work?

airflow google-cloud-dataproc apache-spark google-cloud-platform
1 Answers

This proposed documentation helped me to make it work

https://user-images.githubusercontent.com/16856802/140657172-986e934f-440f-4d1a-882d-14c47541aaa1.png

First you need to import this library:

from airflow.providers.google.cloud.operators.dataproc import ClusterGenerator

And then define the task like it is explained in the image, e.g.

    path = f"gs://goog-dataproc-initialization-actions-us-east4/connectors/connectors.sh"

    CLUSTER_CONFIG =  ClusterGenerator(
        project_id='project-id',
        zone='us-east4-a',
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1',
        num_workers=2,
        init_actions_uris=[
            path
        ],
        image_version='1.5-debian10',
        metadata={'spark-bigquery-connector-url':'gs://spark-lib/bigquery/spark-2.4-bigquery-0.23.1-preview.jar'},
    ).make()

    create_dataproc = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='dataproc-cluster-demo-{{ ds_nodash }}',
        project_id='project-id',
        region='us-east4',
        cluster_config=CLUSTER_CONFIG,
    )
user_154411040
Related