Tfx: Pass the parameters when manually trigger Airflow DAG via CLI?

[LuBingtan]

As I known, we can pass parameters by cli when using airflow. For example:

command line

airflow trigger_dag example_dag --conf '{"parameter":"value"}'

python operator

def print_context(ds, **kwargs):
    parameter = kwargs['dag_run'].conf["parameter"]
    print("received parameter: ", parameter)
    return 'Whatever you return gets printed in the logs'

run_this = PythonOperator(
    task_id='print_the_context',
    provide_context=True,
    python_callable=print_context,
    dag=dag)

But I did not find how to use this feature in TFX. I only found this in souce code(https://github.com/tensorflow/tfx/blob/master/tfx/orchestration/airflow/airflow_component.py)

def _airflow_component_launcher(
    component: base_component.BaseComponent, component_launcher_class: Type[
        base_component_launcher.BaseComponentLauncher],
    pipeline_info: data_types.PipelineInfo, driver_args: data_types.DriverArgs,
    metadata_connection_config: metadata_store_pb2.ConnectionConfig,
    beam_pipeline_args: List[Text], additional_pipeline_args: Dict[Text, Any],
    **kwargs) -> None:
  """Helper function to launch TFX component execution.
  This helper function will be called with Airflow env objects which contains
  run_id that we need to pass into TFX ComponentLauncher.
  Args:
    component: TFX BaseComponent instance. This instance holds all inputs and
      outputs placeholders as well as component properties.
    component_launcher_class: the class of the launcher to launch the component.
    pipeline_info: a data_types.PipelineInfo instance that holds pipeline
      properties
    driver_args: component specific args for driver.
    metadata_connection_config: configuration for how to connect to metadata.
    beam_pipeline_args: Beam pipeline args for beam jobs within executor.
    additional_pipeline_args: a dict of additional pipeline args.
    **kwargs: Context arguments that will be passed in by Airflow, including:
      - ti: TaskInstance object from which we can get run_id of the running
        pipeline.
      For more details, please refer to the code:
      https://github.com/apache/airflow/blob/master/airflow/operators/python_operator.py
  """
  # Populate run id from Airflow task instance.
  pipeline_info.run_id = kwargs['ti'].get_dagrun().run_id
  launcher = component_launcher_class.create(
      component=component,
      pipeline_info=pipeline_info,
      driver_args=driver_args,
      metadata_connection_config=metadata_connection_config,
      beam_pipeline_args=beam_pipeline_args,
      additional_pipeline_args=additional_pipeline_args)
  launcher.launch()

It seemed that kwags is not passed into pipeline args?
So, Is there any way to pass parameters from CLI ?

[ruoyu90]

Hi @LuBingtan , we're actively working on supporting runtime parameter. Leaving this open as a FR.
/cc+ @numerology who is working on that.

[mrcolo]

Any update on this?