Facing Issue with DataprocCreateClusterOperator (Airflow 2.0)

Question

I'm trying to migrate from airflow 1.10 to Airflow 2 which has a change of name for some operators which includes - DataprocClusterCreateOperator. Here is an extract of the code.

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

with dag:

    create_dataproc_cluster = dataproc_operator.DataprocCreateClusterOperator(
        task_id = "create_test_dataproc_cluster",
        project_id="project_id",
        region="us-central1",
        cluster_name="cluster_name",
        tags="dataproc",
        num_workers=2,
        storage_bucket=None,
        num_masters=1,
        master_machine_type="n1-standard-4",
        master_disk_type="pd-standard",
        master_disk_size=500,
        worker_machine_type="n1-standard-4",
        worker_disk_type="pd-standard",
        worker_disk_size=500,
        properties={},
        image_version="1.5-ubuntu18",
        autoscaling_policy=None,
        idle_delete_ttl=7200,
        optional_components=['JUPYTER', 'ANACONDA'],
        metadata={"bigquery-connector-version": '1.1.1',
                  "spark-bigquery-connector-version": '0.17.2',
                  "PIP_PACKAGES" : 'oyaml datalab'},
        init_actions_uris =['gs://goog-dataproc-initialization-actions-us-central1/connectors/connectors.sh','gs://goog-dataproc-initialization-actions-us-central1/python/pip-install.sh']
        )
        
    create_dataproc_cluster

I am ruuning into below error:

File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataproc.py", line 325, in create_cluster
    result = client.create_cluster(
  File "/opt/python3.8/lib/python3.8/site-packages/google/cloud/dataproc_v1beta2/services/cluster_controller/client.py", line 445, in create_cluster
    response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/gapic_v1/method.py", line 145, in __call__
    return wrapped_func(*args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/timeout.py", line 102, in func_with_timeout
    return func(*args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/grpc_helpers.py", line 67, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/grpc/_channel.py", line 944, in __call__
    state, call, = self._blocking(request, timeout, metadata, credentials,
  File "/opt/python3.8/lib/python3.8/site-packages/grpc/_channel.py", line 926, in _blocking
    call = self._channel.segregated_call(
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 498, in grpc._cython.cygrpc.Channel.segregated_call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 366, in grpc._cython.cygrpc._segregated_call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 360, in grpc._cython.cygrpc._segregated_call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 218, in grpc._cython.cygrpc._call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 246, in grpc._cython.cygrpc._call
  File "src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi", line 89, in grpc._cython.cygrpc._operate
  File "src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi", line 64, in grpc._cython.cygrpc._BatchOperationTag.prepare
  File "src/python/grpcio/grpc/_cython/_cygrpc/operation.pyx.pxi", line 37, in grpc._cython.cygrpc.SendInitialMetadataOperation.c
  File "src/python/grpcio/grpc/_cython/_cygrpc/metadata.pyx.pxi", line 41, in grpc._cython.cygrpc._store_c_metadata
ValueError: too many values to unpack (expected 2)

On debugging I see this issue is due to the param metadata. Does anyone has idea what is wrong with this param or a way to fix it.

user_1386551 · Answer

It seems that in this version the type of metadata parameter is no longer dict. From the docs:

metadata (Sequence[Tuple[str, str]]) -- Additional metadata that is provided to the method.

Try with:

metadata = [
    ("bigquery-connector-version", '1.1.1'),
    ("spark-bigquery-connector-version", '0.17.2'),
    ("PIP_PACKAGES", 'oyaml datalab')
]

Edit

According to this issue, you'll need to Generate Cluster Config then pass it to DataprocCreateClusterOperator:

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

CLUSTER_GENERATOR_CONFIG = ClusterGenerator(
    task_id = "create_test_dataproc_cluster",
    project_id="project_id",
    region="us-central1",
    metadata={'PIP_PACKAGES': 'yaml datalab'},
    # ....
).make()

create_cluster_operator = DataprocCreateClusterOperator(
    task_id='create_dataproc_cluster',
    cluster_name="test",
    project_id="project_id",
    region="us-central1",
    cluster_config=CLUSTER_GENERATOR_CONFIG,
)

Facing Issue with DataprocCreateClusterOperator (Airflow 2.0)

1 Answers

Edit