Local mode pdi

[rafahv]

Buenas, Quiero añadir la opcion de poder lanzar pdi en modo local en la misma maquina donde esta airflow, pero tengo varias dudas :

• No se muy bien como haceis para leer la configuacion extra donde estaria el path de pdi o si incluso es necesario.

• Si ejecuto el test TestKitchenOperatorLocal me lanza la etl sin problema , pero me sale este error y no entiendo muy bien el problema ` tests/integration/operators/test_kitchen_local.py:23 (TestKitchenOperatorLocal.test_return_value) self =

  def test_return_value(self): op= KitchenLocalOperator( task_id='test_kitchen_operator', xcom_push=True,

  directory for kitchen

      directory='/Users/rvalenzuela/Documents/data-integration',
      job='/Users/rvalenzuela/Documents/Proyectos/airflow-pentaho-plugin-master/airflow-pentaho-plugin/tests/assets/test_job.kjb',
      params={'a': '1'}
  )

    return_value = op.execute(context={})

test_kitchen_local.py:33:

---

../../../.venv/lib/python3.12/site-packages/airflow/models/baseoperator.py:400: in wrapper return func(self, *args, **kwargs) ../../../airflow_pentaho/operators/kettle.py:414: in execute self.xcom_push(context, key=XCOM_RETURN_KEY, value=output)

---

context = {}, key = 'return_value' value = '2024/05/21 21:50:28 - Kitchen - Processing ended after 0 seconds.' execution_date = None

@staticmethod
def xcom_push(
    context: Any,
    key: str,
    value: Any,
    execution_date: datetime | None = None,
) -> None:
    """
    Make an XCom available for tasks to pull.

    :param context: Execution Context Dictionary
    :param key: A key for the XCom
    :param value: A value for the XCom. The value is pickled and stored
        in the database.
    :param execution_date: if provided, the XCom will not be visible until
        this date. This can be used, for example, to send a message to a
        task on a future date without it being immediately visible.
    """

  context["ti"].xcom_push(key=key, value=value, execution_date=execution_date)

E KeyError: 'ti'

../../../.venv/lib/python3.12/site-packages/airflow/models/baseoperator.py:1564: KeyError

Process finished with exit code 1 `

[sonarcloud[bot]]

Quality Gate failed

Failed conditions
1 Security Hotspot
28.5% Duplication on New Code (required ≤ 3%)

See analysis details on SonarCloud

[piffall]

Hola @rafahv ,

Esta funcionalidad ya esta soportada en el plugin. Lo único que tienes que hacer es configurar la conexión de la siguiente forma (indicando en el Extra, pentaho_home y rep, descarta la configuación para Carte):

https://github.com/damavis/airflow-pentaho-plugin?tab=readme-ov-file#airflow-connection

Cuando tengas la connexión configurada, puedes usar el KitchenOperator o el PanOperator, que ya son para ejecución en local. Recuerda que el componente de Airflow que tiene que estar instalado en el mismo host que Pentaho, es el worker.

Otro detalle, relacionado con tu error, es el uso de xcom_push, que tal vez tengas que ponerlo a False.

Por desgracia tengo que rechazar este PR.

Un saludo.

[rafahv]

Buenas, Pero sino me equivoco tu estas usando en repositorio de pentaho y es una pieza que va desaparecer y quería llamar directamente a data-integration , quizá me equivoque.

De: Cristòfol Torrens @.> Fecha: miércoles, 22 de mayo de 2024, 7:54 Para: damavis/airflow-pentaho-plugin @.> CC: Rafael Valenzuela @.>, Mention @.> Asunto: Re: [damavis/airflow-pentaho-plugin] Local mode pdi (PR #35) EXTERNAL EMAIL

Hola @rafahvhttps://github.com/rafahv ,

Esta funcionalidad ya esta soportada en el plugin. Lo único que tienes que hacer es configurar la conexión de la siguiente forma (indicando en el Extra, pentaho_home y rep, descarta la configuación para Carte):

https://github.com/damavis/airflow-pentaho-plugin?tab=readme-ov-file#airflow-connection

Cuando tengas la connexión configurada, puedes usar el KitchenOperator o el PanOperator, que ya son para ejecución en local. Recuerda que el componente de Airflow que tiene que estar instalado en el mismo host que Pentaho, es el worker.

Por desgracia tengo que rechazar este PR.

Un saludo.

— Reply to this email directly, view it on GitHubhttps://github.com/damavis/airflow-pentaho-plugin/pull/35#issuecomment-2123918897, or unsubscribehttps://github.com/notifications/unsubscribe-auth/BBBNCZQCGNDK7B3QW5QWPZDZDQXHRAVCNFSM6AAAAABICIME2WVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDCMRTHEYTQOBZG4. You are receiving this because you were mentioned.Message ID: @.***>

[piffall]

Hola @rafahv KitcheckOperator implementa la llamada al script de PDI llamado kitchen.sh. Si puedes ejecutar tu Job con ese script, también podrás hacerlo con KitchenOperator.

Debería funcionar simplemente con este código:

job = KitchenOperator(
    conn_id='pdi_default',
    task_id='test_job',
    job='/Users/rvalenzuela/Documents/Proyectos/airflow-pentaho-plugin-master/airflow-pentaho-plugin/tests/assets/test_job.kjb',
    params={'a': '1'},
    dag=dag)

[rafahv]

gracias