How to pass all dag_run.conf parameters to python_wheel_task

Bartek — Thu, 19 Jan 2023 21:02:49 GMT

I want to trigger Databricks job from Airflow using DatabricksSubmitRunDeferrableOperator and I need to pass configuration params. Here is excerpt from my code (definition is not complete, only crucial properties):

from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunDeferrableOperator
 
 
class PythonWheelOperator(DatabricksSubmitRunDeferrableOperator):
   def __init__(self, action_name, **kwargs):
       DatabricksSubmitRunDeferrableOperator.__init__(self,
                                                      databricks_conn_id=dag_config_json["databricksConnId"],
                                                      task_id=action_name,
                                                      tasks=[
                                                           {
                                                               "python_wheel_task": {
                                                                   "package_name": "PName",
                                                                   "entry_point": "ReviewSuppliers",
                                                                   "named_parameters": "{{dict(dag_run.conf)}}"
                                                               },
                                                               "libraries": [
                                                                   {
                                                                       "whl": "dbfs:/FileStore/path-to-whl.whl"
                                                                   }
                                                      )

Error I got:

airflow.exceptions.AirflowException: Response: b'{"error_code":"MALFORMED_REQUEST","message":"Could not parse request object: Expected \'START_OBJECT\' not \'VALUE_STRING\'\\n at [Source: (ByteArrayInputStream); line: 1, column: 187]\\n at [Source: java.io.ByteArrayInputStream@47fb05c6; line: 1, column: 187]"}', Status Code: 400

How should I change line: "named_parameters": "{{dict(dag_run.conf)}}" to properly pass this config parameters?

topic How to pass all dag_run.conf parameters to python_wheel_task in Data Engineering

How to pass all dag_run.conf parameters to python_wheel_task