Der Composer-Workflow schlägt beim Dataproc-Operator fehl
Ich habe eine Composer-Umgebung in gcp eingerichtet und es wird eine DAG wie folgt ausgeführt
with DAG('sample-dataproc-dag',
default_args=DEFAULT_DAG_ARGS,
schedule_interval=None) as dag: # Here we are using dag as context
# Submit the PySpark job.
submit_pyspark = DataProcPySparkOperator(
task_id='run_dataproc_pyspark',
main='gs://.../dataprocjob.py',
cluster_name='xyz',
dataproc_pyspark_jars=
'gs://.../spark-bigquery-latest_2.12.jar'
)
simple_bash = BashOperator(
task_id='simple-bash',
bash_command="ls -la")
submit_pyspark.dag = dag
submit_pyspark.set_upstream(simple_bash)
und das ist meine dataprocjob.py
from pyspark.sql import SparkSession
if __name__ == '__main__':
spark = SparkSession.builder.appName('Jupyter BigQuery Storage').getOrCreate()
table = "projct.dataset.txn_w_ah_demo"
df = spark.read.format("bigquery").option("table",table).load()
df.printSchema()
Meine Composer-Pipeline schlägt beim Dataproc-Schritt fehl. In dem im gcs gespeicherten Komponistenprotokoll sehe ich Folgendes
[2020-09-23 21:40:02,849] {taskinstance.py:1059} ERROR - <HttpError 403 when requesting https://dataproc.googleapis.com/v1beta2/projects/lt-dia-pop-dis-upr/regions/global/jobs?clusterName=dppoppr004&alt=json returned "Not authorized to requested resource.">@-@{"workflow": "sample-dataproc-dag", "task-id": "run_dataproc_pyspark", "execution-date": "2020-09-23T21:39:42.371933+00:00"}
Traceback (most recent call last):
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 930, in _run_raw_task
result = task_copy.execute(context=context)
File "/usr/local/lib/airflow/airflow/contrib/operators/dataproc_operator.py", line 1139, in execute
super(DataProcPySparkOperator, self).execute(context)
File "/usr/local/lib/airflow/airflow/contrib/operators/dataproc_operator.py", line 707, in execute
self.hook.submit(self.hook.project_id, self.job, self.region, self.job_error_states)
File "/usr/local/lib/airflow/airflow/contrib/hooks/gcp_dataproc_hook.py", line 311, in submit
num_retries=self.num_retries)
File "/usr/local/lib/airflow/airflow/contrib/hooks/gcp_dataproc_hook.py", line 51, in __init__
clusterName=cluster_name).execute()
File "/opt/python3.6/lib/python3.6/site-packages/googleapiclient/_helpers.py", line 130, in positional_wrapper
return wrapped(*args, **kwargs)
File "/opt/python3.6/lib/python3.6/site-packages/googleapiclient/http.py", line 851, in execute
raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 403 when requesting https://dataproc.googleapis.com/v1beta2/projects/lt-dia-pop-dis-upr/regions/global/jobs?clusterName=dppoppr004&alt=json returned "Not authorized to requested resource.">
Antworten
Es scheint, dass das von Ihnen präsentierte Problem den Dataproc-Berechtigungen entspricht, die Sie Ihrer Anwendung erteilt haben.
Laut Dokumentation benötigen Sie verschiedene Rollenberechtigungen , um Dataproc-Aufgaben auszuführen, zum Beispiel:
dataproc.clusters.create permits the creation of Cloud Dataproc clusters in the containing project
dataproc.jobs.create permits the submission of Dataproc jobs to Dataproc clusters in the containing project
dataproc.clusters.list permits the listing of details of Dataproc clusters in the containing project
Wenn Sie einen Dataproc-Auftrag senden möchten, benötigen Sie die Berechtigungen 'dataproc.clusters.use' und 'dataproc.jobs.create'.
Um Ihrem Benutzerkonto die richtigen Berechtigungen zu erteilen, können Sie der Dokumentation zum Aktualisieren Ihres Dienstkontos, das Sie in Ihrem Code verwenden, folgen und die richtigen Berechtigungen hinzufügen.
Auf den ersten Blick sieht es so aus, als ob die Berechtigungen für das Google Cloud-Konto, mit denen Sie die Dataproc-APIs aufrufen, für den Bediener nicht ausreichend sind.