à¨à© à¨¹à© dag_run.conf à¨²à¨ à¨µà¨°à¨¤à¨¿à¨?

à¨µà¨°à¨¤à¨£ à¨¦à¨¾ à¨à© à¨²à¨¾à¨ à¨¹à© BranchPythonOperator?

ਡੈਗ ਰਨ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੀ

Alice Dupont

ਵੀਰਵਾਰ, 13 ਫ਼ਰਵਰੀ 2025 7:44:55 ਬਾ.ਦੁ.

ਏਅਰਫਲੋ ਵਿਚ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਨਿਰਭਰਤਾ ਦੀ ਸ਼ਕਤੀ ਨੂੰ ਤਾਲਾ ਖੋਲ੍ਹਣਾ

ਅਪਾਚੇ ਏਅਰਫਲੋ ਇੱਕ ਸ਼ਕਤੀਸ਼ਾਲੀ ਵਰਕਫਲੋਅ ਆਟੋਮੈਟਿਕ ਟੂਲ ਹੈ, ਪਰ ਗਤੀਸ਼ੀਲ ਨਿਰਭਰਤਾ ਨੂੰ ਸੰਭਾਲਣਾ ਕਈ ਵਾਰ ਬੁਝਾਰਤ ਨੂੰ ਹੱਲ ਕਰਨ ਵਾਂਗ ਮਹਿਸੂਸ ਕਰ ਸਕਦਾ ਹੈ. ਇੱਕ ਨਿਰਦੇਸ਼ਿਤ ਐਕਸੀਲ ਗ੍ਰਾਫ (ਡੀਏਜੀ), ਹਾਰਡ ਕੋਡਿੰਗ ਟਾਸਕ ਦੇ ਕ੍ਰਮ ਸਧਾਰਣ ਵਰਤੋਂ ਦੇ ਸਧਾਰਣ ਮਾਮਲਿਆਂ ਲਈ ਕੰਮ ਕਰ ਸਕਦੇ ਹਨ, ਪਰ ਉਦੋਂ ਕੀ ਜੇ ruct ਾਂਚੇ ਨੂੰ ਰਨਟਾਈਮ ਤੇ ਨਿਸ਼ਚਤ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੁੰਦੀ ਹੈ? 🤔

ਕਲਪਨਾ ਕਰੋ ਕਿ ਤੁਸੀਂ ਇੱਕ ਡੇਟਾ ਪਾਈਪਲਾਈਨ 'ਤੇ ਕੰਮ ਕਰ ਰਹੇ ਹੋ ਜਿਥੇ ਆਉਣ ਵਾਲੇ ਡੇਟਾ ਤੇ ਨਿਰਭਰ ਕਰਦਾ ਹੈ. ਉਦਾਹਰਣ ਦੇ ਲਈ, ਰੋਜ਼ਾਨਾ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੇ ਅਧਾਰ ਤੇ ਫਾਈਲਾਂ ਦੇ ਵੱਖ ਵੱਖ ਸੈੱਟਾਂ ਤੇ ਕਾਰਵਾਈ ਕਰਨਾ ਜਾਂ ਵਪਾਰਕ ਨਿਯਮ ਦੇ ਅਧਾਰ ਤੇ ਪਰਿਵਰਤਨਸ਼ੀਲ ਰੂਪਾਂਤਰਣ ਨੂੰ ਚਲਾਉਣ. ਅਜਿਹੇ ਮਾਮਲਿਆਂ ਵਿੱਚ, ਇੱਕ ਸਥਿਰ ਡੀਗ ਇਸਨੂੰ ਨਹੀਂ ਵੱ cut ਨਹੀਂ ਦੇਵੇਗਾ - ਤੁਹਾਨੂੰ ਗਤੀਸ਼ੀਲਤਾ ਨੂੰ ਪ੍ਰਭਾਸ਼ਿਤ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੈ.

ਇਹ ਬਿਲਕੁਲ ਉਹੀ ਹੈ ਜਿਥੇ ਏਅਰਫਲੋਅ ਹੈ dag_run.conf ਖੇਡ-ਚੇਂਜਰ ਹੋ ਸਕਦਾ ਹੈ. ਇੱਕ DAG ਚਾਲੂ ਕਰਨ ਵੇਲੇ ਇੱਕ ਕੌਂਫਿਗਰੇਸ਼ਨ ਸ਼ਬਦਕੋਸ਼ ਪਾਸ ਕਰਕੇ, ਤੁਸੀਂ ਆਰਜੀ ਤੌਰ ਤੇ ਕੰਮ ਦੇ ਕ੍ਰਮ ਤਿਆਰ ਕਰ ਸਕਦੇ ਹੋ. ਹਾਲਾਂਕਿ, ਇਸ ਨੂੰ ਇੱਕ struct ਾਂਚਾਗਤ in ੰਗ ਨਾਲ ਲਾਗੂ ਕਰਨਾ ਏਅਰਫਲੋ ਦੇ ਐਗਜ਼ੀਕਿ .ਸ਼ਨ ਮਾਡਲ ਦੀ ਡੂੰਘੀ ਸਮਝ ਦੀ ਲੋੜ ਹੈ.

ਇਸ ਲੇਖ ਵਿਚ, ਅਸੀਂ ਪੜਚੋਲ ਕਰਾਂਗੇ ਕਿ ਗਤੀਸ਼ੀਲ ਡੱਗ ਕਿਵੇਂ ਬਣਾਇਆ ਜਾਵੇ ਜਿਥੇ ਟੌਸਟ ਦੀ ਨਿਰਭਰਤਾ ਰਨਟਾਈਮ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਨਿਰਧਾਰਤ ਕੀਤੀ ਜਾਂਦੀ ਹੈ dag_run.conf. ਜੇ ਤੁਸੀਂ ਇਸ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਸੰਘਰਸ਼ ਕਰ ਰਹੇ ਹੋ ਅਤੇ ਸਪਸ਼ਟ ਹੱਲ ਨਹੀਂ ਲੱਭਿਆ, ਤਾਂ ਚਿੰਤਾ ਨਾ ਕਰੋ - ਤੁਸੀਂ ਇਕੱਲੇ ਨਹੀਂ ਹੋ! ਆਓ ਆਪਾਂ ਉਦਾਹਰਣਾਂ ਨਾਲ ਕਦਮ ਨਾਲ ਕਦਮ ਨਾਲ ਕਦਮ ਰੱਖੀਏ. 🚀

ਕਮਾਂਡ	ਵਰਤੋਂ ਦੀ ਉਦਾਹਰਣ
dag_run.conf	ਡੀਗ ਰਨ ਨੂੰ ਚਾਲੂ ਕਰਨ ਵੇਲੇ ਗਤੀਸ਼ੀਲ ਸੰਰਚਨਾ ਮੁੱਲ ਪ੍ਰਾਪਤ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ. ਰੰਨਟਾਈਮ ਪੈਰਾਮੀਟਰ ਲੰਘਣ ਲਈ ਜ਼ਰੂਰੀ.
PythonOperator	ਏਅਰਫਲੋ ਵਿਚ ਇਕ ਕੰਮ ਦੀ ਪਰਿਭਾਸ਼ਾ ਦਿੰਦਾ ਹੈ ਜੋ ਇਕ ਪਾਇਥਨ ਫੰਕਸ਼ਨ ਨੂੰ ਚਲਾਉਂਦਾ ਹੈ, ਜਿਸ ਨਾਲ ਡੱਗ ਦੇ ਅੰਦਰ ਲਚਕਦਾਰ ਐਗਜ਼ੀਕਿ .ਸ਼ਨ ਲੌਜਿਕ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ.
set_upstream()	ਸਪੱਸ਼ਟ ਤੌਰ 'ਤੇ ਕਾਰਜਾਂ ਵਿਚਕਾਰ ਨਿਰਭਰਤਾ ਨੂੰ ਪ੍ਰਭਾਸ਼ਿਤ ਕਰਦਾ ਹੈ, ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਦੇ ਹਨ ਕਿ ਇਕ ਹੋਰ ਤੋਂ ਬਾਅਦ ਇਕ ਕੰਮ ਪੂਰਾ ਹੋ ਗਿਆ ਹੈ.
@dag	ਇਕ ਪਨੀਰ-ਬਣਤਰ ਅਤੇ struct ਾਂਚਾਗਤ in ੰਗ ਨਾਲ ਡਗਾਂ ਨੂੰ ਪ੍ਰਭਾਸ਼ਿਤ ਕਰਨ ਲਈ ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਦੁਆਰਾ ਪ੍ਰਦਾਨ ਕੀਤਾ ਜਾਂਦਾ ਹੈ.
@task	ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਏਅਰਫਲੋ ਦੀ ਪਰਿਭਾਸ਼ਾ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ, ਸਰਲਾਈਜ਼ਿੰਗ ਟਾਸਕ ਰਚਨਾ ਅਤੇ ਡੇਟਾ ਪਾਸ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ.
override(task_id=...)	ਇਕਲੌਇਂਸ ਨੂੰ ਇਕੋ ਫੰਕਸ਼ਨ ਤੋਂ ਕਈ ਕਾਰਜਾਂ ਨੂੰ ਸਥਾਪਤ ਕਰਨ ਵੇਲੇ ਇਕ ਕੰਮ ਦੀ ਆਈਡੀ ਨੂੰ ਸੋਧਣ ਲਈ ਵਰਤਿਆ ਜਾਂਦਾ ਹੈ.
extract_elements(dag_run=None)	ਇੱਕ ਫੰਕਸ਼ਨ ਜੋ dag_run.conf ਕੋਸ਼ਾਂ ਤੋਂ ਮੁੱਲ ਲੈਂਦਾ ਹੈ ਡਿਕਸ਼ਨਰੀ ਵਿੱਚ ਟਾਸਕ ਫਾਂਸੀ ਨੂੰ ਸੰਰਚਿਤ ਕਰਨ ਲਈ.
schedule_interval=None	ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਦਾ ਹੈ ਕਿ ਡੀਏਜੀ ਨੂੰ ਸਿਰਫ ਇੱਕ ਨਿਸ਼ਚਤ ਸ਼ਡਿ .ਲ ਤੇ ਚੱਲਣ ਦੀ ਬਜਾਏ ਸਿਰਫ DAG ਚਲਾਇਆ ਜਾਂਦਾ ਹੈ.
op_args=[element]	ਇੱਕ ਪਾਈਥਨਓਪਰਸਟਰਸ ਕੰਮ ਨੂੰ ਗਤੀਸ਼ੀਲ ਆਰਗੂਮੈਂਟਸ ਪਾਸ ਕਰਦਾ ਹੈ, ਪ੍ਰਤੀ ਕਾਰਜ ਦੀ ਵੱਖ ਵੱਖ ਕਾਰਜਾਂ ਨੂੰ ਯੋਗ ਕਰਦਾ ਹੈ.
catchup=False	ਵਿਰਾਮ ਦੇ ਬਾਅਦ ਸ਼ੁਰੂ ਹੋਣ ਤੇ ਏਅਰਫਲੋ ਨੂੰ ਸਾਰੇ ਖੁੰਝਾਉਣ ਤੋਂ ਬਾਅਦ ਬੰਦ ਕਰਨ ਤੋਂ ਰੋਕਦਾ ਹੈ, ਰੀਅਲ-ਟਾਈਮ ਕੌਂਫਿਗਰੇਸ਼ਨਾਂ ਲਈ ਲਾਭਦਾਇਕ ਹੋਣ.

ਏਅਰਫਲੋ ਵਿੱਚ ਰੰਨਟਾਈਮ ਕੌਂਫਿਗਰੇਸ਼ਨ ਨਾਲ ਗਤੀਸ਼ੀਲ ਡੈਗ ਬਣਾਉਣਾ

ਅਪਾਚੇ ਏਅਰਫਲੋ ਗੁੰਝਲਦਾਰ ਕਾਰਜ ਮੁਹੈਚ ਨੂੰ ਰੋਕਦਾ ਹੈ, ਪਰ ਇਸਦੀ ਸੱਚੀ ਤਾਕਤ ਇਸ ਦੀ ਲਚਕਤਾ ਵਿੱਚ ਹੈ. ਸਕ੍ਰਿਪਟਾਂ ਨੇ ਪਹਿਲਾਂ ਦਿਖਾਇਆ ਕਿ ਕਿਵੇਂ ਬਣਾਏ ਜਾ ਸਕਦੇ ਹਨ ਡਾਇਨਾਮਿਕ ਡੈਗ ਜਿੱਥੇ ਟੌਸਟਿਸ ਨਿਰਭਰਤਾ ਰਨਟਾਈਮ ਦੀ ਵਰਤੋਂ ਕੀਤੀ ਜਾਂਦੀ ਹੈ dag_run.conf. ਪ੍ਰਕਿਰਿਆਵਾਂ ਦੇ ਤੱਤਾਂ ਦੀ ਸੂਚੀ ਨੂੰ ਘਟਾਉਣ ਦੀ ਬਜਾਏ, ਜਦੋਂ ਚਾਲੂ ਹੋਣ ਤੇ, ਤਾਂ ਡਾਇਗਲ ਨੂੰ ਗਤੀਸ਼ੀਲਤਾ ਨਾਲ ਮੁੜ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹੈ. ਇਹ ਵਿਸ਼ੇਸ਼ ਤੌਰ 'ਤੇ ਅਸਲ-ਸੰਸਾਰ ਦੇ ਦ੍ਰਿਸ਼ਾਂ ਵਿੱਚ ਲਾਭਦਾਇਕ ਹੈ, ਜਿਵੇਂ ਕਿ ਪਰਿਵਰਤਨਸ਼ੀਲ ਡੇਟਾਨੇਟਸ ਜਾਂ ਬਾਹਰੀ ਸਥਿਤੀਆਂ ਦੇ ਅਧਾਰ ਤੇ ਵਿਸ਼ੇਸ਼ ਕਾਰਜਾਂ ਨੂੰ ਚਲਾਉਣ. ਇਕ ਈਟੀਐਲ ਪਾਈਪਲਾਈਨ ਦੀ ਕਲਪਨਾ ਕਰੋ ਜਿੱਥੇ ਰੋਜ਼ਾਨਾ-ਇਹ ਪਹੁੰਚ ਸਵੈਚਾਲਨ ਨੂੰ ਬਹੁਤ ਸੌਖਾ ਬਣਾਉਂਦੀ ਹੈ. 🚀

ਪਹਿਲੀ ਸਕ੍ਰਿਪਟ ਇਸ ਦੀ ਵਰਤੋਂ ਕਰਦੀ ਹੈ ਪਾਈਥਨਓਪਰੇਟੋਰ ਕਾਰਜਾਂ ਨੂੰ ਲਾਗੂ ਕਰਨ ਅਤੇ ਨਿਰਭਰਤਾ ਨੂੰ ਆਰਜੀ ਤੌਰ ਤੇ ਨਿਰਧਾਰਤ ਕਰਨ ਲਈ. ਇਹ ਤੱਤ ਦੀ ਸੂਚੀ ਕੱ racts ਸਕਦੀ ਹੈ dag_run.confਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਨਾ ਕਿ ਕੰਮ ਉਦੋਂ ਹੀ ਬਣਾਏ ਜਾਂਦੇ ਹਨ ਜਦੋਂ ਲੋੜ ਹੋਵੇ. ਸੂਚੀ ਵਿੱਚ ਹਰੇਕ ਤੱਤ ਇੱਕ ਵਿਲੱਖਣ ਕੰਮ ਬਣ ਜਾਂਦਾ ਹੈ, ਅਤੇ ਨਿਰਭਰਤਾ ਕ੍ਰਮਵਾਰ ਬਣ ਜਾਂਦੀ ਹੈ. ਦੂਜੀ ਪਹੁੰਚ ਦਾ ਲਾਭ ਉਠਾਓ ਟਾਸਕਫਲੋ ਏਪੀਆਈ, ਜੋ ਕਿ ਡੱਗ ਰਚਨਾ ਨੂੰ ਸਜਾਵਟ ਨਾਲ ਜੋੜਦੀ ਹੈ @ਡੀਗ ਅਤੇ @task. ਇਹ ਵਿਧੀ dag ਨੂੰ ਵਧੇਰੇ ਪੜ੍ਹਨਯੋਗ ਬਣਾਉਂਦੀ ਹੈ ਅਤੇ ਕਲੀਨਰ ਫਾਂਸੀ ਦੇ ਤਰਕ ਨੂੰ ਬਣਾਈ ਰੱਖਦੀ ਹੈ. ਇਹ ਪਹੁੰਚ ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਦੇ ਹਨ ਕਿ ਵਰਕਫਲੋਜ਼ ਬਿਨਾਂ ਕੋਡ ਤਬਦੀਲੀਆਂ ਦੀ ਜ਼ਰੂਰਤ ਤੋਂ ਬਿਨਾਂ ਵੱਖ ਵੱਖ ਕੌਂਫਿਗਰੇਸ਼ਨਾਂ ਤੇ .ਾਲ ਸਕਦੇ ਹਨ.

ਉਦਾਹਰਣ ਦੇ ਲਈ, ਇੱਕ ਦ੍ਰਿਸ਼ 'ਤੇ ਗੌਰ ਕਰੋ ਜਿੱਥੇ ਇਕ ਈ-ਕਾਮਰਸ ਕੰਪਨੀ ਬੈਚਾਂ ਵਿਚ ਆਰਡਰ ਕਰਦੀ ਹੈ. ਕੁਝ ਦਿਨਾਂ ਵਿੱਚ ਦੂਜਿਆਂ ਨਾਲੋਂ ਜ਼ਿਆਦਾ ਜ਼ਰੂਰੀ ਆਰਡਰ ਹੋ ਸਕਦੇ ਹਨ, ਵੱਖਰੇ ਕੰਮ ਦੇ ਕ੍ਰਮਾਂ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ. ਸਥਿਰ ਡੀਏਜੀ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦਾ ਅਰਥ ਹੈ ਕੋਡ ਨੂੰ ਹਰ ਵਾਰ ਪ੍ਰਾਈਵੇਟ ਬਦਲੋ. ਸਾਡੀ ਗਤੀਸ਼ੀਲ ਡੀਆਈਜੀ ਪਹੁੰਚ ਦੇ ਨਾਲ, ਇੱਕ ਬਾਹਰੀ ਸਿਸਟਮ ਇੱਕ ਖਾਸ ਕਾਰਜ ਕ੍ਰਮ ਨਾਲ ਡੱਗ ਨੂੰ ਚਾਲੂ ਕਰ ਸਕਦਾ ਹੈ, ਜਿਸ ਨਾਲ ਪ੍ਰਕਿਰਿਆ ਨੂੰ ਵਧੇਰੇ ਕੁਸ਼ਲ ਬਣਾਉਂਦਾ ਹੈ. ਇਕ ਹੋਰ ਵਰਤੋਂ ਦਾ ਕੇਸ ਡੇਟਾ ਵਿਗਿਆਨ ਵਿਚ ਹੈ, ਜਿੱਥੇ ਮਾਡਲਾਂ ਨੂੰ ਆਉਣ ਵਾਲੀਆਂ ਡੇਟਾ ਡਿਸਟਰੀਬਿ .ਸ਼ਨਾਂ ਦੇ ਅਧਾਰ ਤੇ ਮੁੜ ਸਿਖਲਾਈ ਦੇਣ ਦੀ ਜ਼ਰੂਰਤ ਹੋ ਸਕਦੀ ਹੈ. ਆਰਜੀ ਤੌਰ 'ਤੇ ਲੋੜੀਂਦੀਆਂ ਮਾੱਡੀਆਂ ਨੂੰ ਪਾਸ ਕਰਕੇ, ਸਿਰਫ ਜ਼ਰੂਰੀ ਗਣਨਾ ਨੂੰ ਲਾਗੂ ਕਰ ਦਿੱਤਾ ਜਾਂਦਾ ਹੈ, ਸਮਾਂ ਅਤੇ ਸਰੋਤ ਬਚਾਉਣ. 🎯

ਸੰਖੇਪ ਵਿੱਚ, ਇਹ ਸਕ੍ਰਿਪਟਾਂ ਰਨਟਾਈਮ ਇਨਪੁਟਸ ਦੇ ਅਧਾਰ ਤੇ ਆਰਜੀ ਤੌਰ ਤੇ ਤਿਆਰ ਕਰਨ ਲਈ ਇੱਕ ਨੀਂਹ ਪ੍ਰਦਾਨ ਕਰਦੀਆਂ ਹਨ. ਜ਼ੈਜੂਰੀ ਨਾਲ ਏਅਰਫਲੋ ਦਾ ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਜਾਂ ਰਵਾਇਤੀ ਪਾਈਥੋਨੌਜੋਰਸ ਪਹੁੰਚ, ਡਿਵੈਲਪਰ ਲਚਕਦਾਰ, ਮਾਡਯੂਲਰ ਅਤੇ ਕੁਸ਼ਲ ਵਰਕਫਲੋਜ਼ ਬਣਾ ਸਕਦੇ ਹਨ. ਇਹ ਹੱਥੀਂ ਦਖਲ ਦੀ ਜ਼ਰੂਰਤ ਨੂੰ ਖਤਮ ਕਰਦਾ ਹੈ ਅਤੇ ਦੂਜੇ ਸਵੈਚਾਲਨ ਪ੍ਰਣਾਲੀਆਂ ਨਾਲ ਸਹਿਜ ਏਕੀਕਰਣ ਲਈ ਆਗਿਆ ਦਿੰਦਾ ਹੈ. ਕੀ ਗਾਹਕ ਦੇ ਆਦੇਸ਼ਾਂ ਤੇ ਕਾਰਵਾਈ ਕਰਦਿਆਂ, ਬੱਦਲ ਦੇ ਵਰਕਫਲੋਜ਼, ਡਾਇਨਾਮਿਕ ਡੈਗਾਂ ਦੇ ਪ੍ਰਬੰਧਨ ਵਿੱਚ, ਗਤੀਸ਼ੀਲ ਡੱਗਸ ਨੂੰ ਖਾਸ ਵਪਾਰਕ ਜ਼ਰੂਰਤਾਂ ਦੇ ਅਨੁਸਾਰ ਸਮਰੱਥ ਬਣਾ ਸਕਦੇ ਹਨ.

ਰਨਟਾਈਮ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੇ ਨਾਲ ਏਅਰਫਲੋ ਵਿੱਚ ਹਵਾ ਦੇ ਪ੍ਰਵਾਹ ਵਿੱਚ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਨੂੰ ਲਾਗੂ ਕਰਨਾ

ਪਾਇਥਨ-ਅਧਾਰਤ ਬੈਕਐਂਡ ਆਟੋਮਲੋਅ ਅਪਾਚੇ ਏਅਰਫਲੋ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.models import DagRun
import json
# Define default args
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(1),
}
# Function to process each element
def process_element(element, kwargs):
    print(f"Processing element: {element}")
# Define DAG
dag = DAG(
    'dynamic_task_dag',
    default_args=default_args,
    schedule_interval=None,
)
# Extract elements from dag_run.conf
def generate_tasks(kwargs):
    conf = kwargs.get('dag_run').conf or {}
    elements = conf.get('elements', [])
    task_list = []
    for i, group in enumerate(elements):
        for j, element in enumerate(group):
            task_id = f"process_element_{i}_{j}"
            task = PythonOperator(
                task_id=task_id,
                python_callable=process_element,
                op_args=[element],
                dag=dag,
            )
            task_list.append(task)
    return task_list
# Generate dynamic tasks
tasks = generate_tasks()
# Define dependencies dynamically
for i in range(len(tasks) - 1):
    tasks[i + 1].set_upstream(tasks[i])

ਵਿਕਲਪਕ ਪਹੁੰਚ: ਬਿਹਤਰ ਪੜ੍ਹਨਯੋਗਤਾ ਲਈ ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਦੀ ਵਰਤੋਂ ਕਰਨਾ

ਆਧੁਨਿਕ ਪਾਈਥਨ ਏਅਰਫਲੋ ਦੇ ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਦੀ ਵਰਤੋਂ ਕਰਦਿਆਂ ਪਹੁੰਚ

from airflow.decorators import dag, task
from datetime import datetime
# Define DAG
@dag(schedule_interval=None, start_date=datetime(2025, 1, 28), catchup=False)
def dynamic_taskflow_dag():
    @task
    def process_element(element: str):
        print(f"Processing {element}")
    @task
    def extract_elements(dag_run=None):
        conf = dag_run.conf or {}
        return conf.get('elements', [])
    elements = extract_elements()
    task_groups = [[process_element(element) for element in group] for group in elements]
    # Define dependencies dynamically
    for i in range(len(task_groups) - 1):
        for upstream_task in task_groups[i]:
            for downstream_task in task_groups[i + 1]:
                downstream_task.set_upstream(upstream_task)
dynamic_taskflow_dag()

ਏਅਰਫਲੋ ਵਿੱਚ ਸ਼ਰਤ-ਨਿਰਦੇਸ਼ਕ ਦੇ ਨਾਲ ਐਕਸ਼ਨਲ ਐਗਜ਼ੀਕਿ .ਸ਼ਨ ਦੇ ਨਾਲ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਨੂੰ ਵਧਾਉਣਾ

ਇੱਕ ਸ਼ਕਤੀਸ਼ਾਲੀ ਹਾਲੇ ਵਿੱਚ ਅਕਸਰ ਅਣਦੇਖੀ ਵਿਸ਼ੇਸ਼ਤਾ ਅਪਾਚੇ ਏਅਰਫਲੋ ਕੰਡੀਸ਼ਨਲ ਫਾਂਸੀ ਦੀ ਹੈ, ਜੋ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਦੀ ਲਚਕਤਾ ਨੂੰ ਵਧਾਉਣ ਲਈ ਹੋਰ ਸੁਧਾਰ ਕਰ ਸਕਦੀ ਹੈ. ਤੋਂ ਟਾਸਕ ਨਿਰਭਰਤਾ ਪ੍ਰਾਪਤ ਕਰਦੇ ਹੋਏ dag_run.conf ਲਾਭਦਾਇਕ ਹੈ, ਅਸਲ-ਸੰਸਾਰ ਦੇ ਦ੍ਰਿਸ਼ਾਂ ਨੂੰ ਅਕਸਰ ਖਾਸ ਸਥਿਤੀਆਂ ਦੇ ਅਧਾਰ ਤੇ ਸਿਰਫ ਕੁਝ ਖਾਸ ਕੰਮਾਂ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ. ਉਦਾਹਰਣ ਦੇ ਲਈ, ਕੁਝ ਡੇਟਾਸੇਟਾਂ ਨੂੰ ਵਿਸ਼ਲੇਸ਼ਣ ਤੋਂ ਪਹਿਲਾਂ ਪ੍ਰੀਪ੍ਰੋਸੈਸਿੰਗ ਦੀ ਜ਼ਰੂਰਤ ਹੋ ਸਕਦੀ ਹੈ, ਜਦੋਂ ਕਿ ਕੁਝ ਸਿੱਧੇ ਤੇ ਕਾਰਵਾਈ ਕੀਤੀ ਜਾ ਸਕਦੀ ਹੈ.

ਹਵਾ ਦੇ ਪ੍ਰਵਾਹ ਵਿੱਚ ਕੰਡੀਸ਼ਨਲ ਐਗਜ਼ੀਕਿ .ਸ਼ਨ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਲਾਗੂ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ BranchPythonOperatorਪਰ, ਪਰਿਭਾਸ਼ਿਤ ਤਰਕ ਦੇ ਅਧਾਰ ਤੇ ਚਲਾਉਣ ਲਈ ਅਗਲਾ ਕੰਮ ਨਿਰਧਾਰਤ ਕਰਦਾ ਹੈ. ਮੰਨ ਲਓ ਕਿ ਸਾਡੇ ਕੋਲ ਇੱਕ ਗਤੀਸ਼ੀਲ ਡੱਗ ਹੈ ਜੋ ਫਾਈਲਾਂ ਤੇ ਕਾਰਵਾਈ ਕਰਦਾ ਹੈ, ਪਰ ਇੱਕ ਅਕਾਰ ਤੋਂ ਉੱਪਰ ਸਿਰਫ ਫਾਈਲਾਂ ਨੂੰ ਪ੍ਰਮਾਣਿਕਤਾ ਦੀ ਲੋੜ ਹੈ. ਸਾਰੇ ਕਾਰਜਾਂ ਨੂੰ ਕ੍ਰਮਵਾਰ ਚਲਾਉਣ ਦੀ ਬਜਾਏ, ਅਸੀਂ ਆਰਜੀ ਤੌਰ ਤੇ ਇਹ ਫੈਸਲਾ ਕਰ ਸਕਦੇ ਹਾਂ ਕਿ ਕਿਹੜੇ ਕੰਮ ਚਲਾਉਣਾ, ਐਗਜ਼ੀਕਿ .ਸ਼ਨ ਦੇ ਸਮੇਂ ਨੂੰ ਅਨੁਕੂਲ ਬਣਾਉਣਾ ਅਤੇ ਸਰੋਤ ਦੀ ਵਰਤੋਂ ਨੂੰ ਘਟਾਉਣਾ ਹੈ. ਇਹ ਪਹੁੰਚ ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਦੀ ਹੈ ਕਿ ਸਿਰਫ attrails ੁਕਵੇਂ ਕੰਮ ਦੇ ਫਲਿਫਲੋਅ ਕੀਤੇ ਜਾਂਦੇ ਹਨ, ਡਾਟਾ ਪਾਈਪਲਾਈਟਸ ਵਧੇਰੇ ਕੁਸ਼ਲ ਬਣਾਉਂਦੇ ਹਨ. 🚀

ਗਤੀਸ਼ੀਲ ਡੱਗ ਨੂੰ ਵਧਾਉਣ ਦਾ ਇਕ ਹੋਰ ਤਰੀਕਾ ਸ਼ਾਮਲ ਕਰਕੇ ਹੈ XComs (ਕਰਾਸ-ਸੰਚਾਰ ਸੰਦੇਸ਼). Xcoms ਕਾਰਜਾਂ ਨੂੰ ਡੇਟਾ ਐਕਸਚੇਜ਼ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ, ਭਾਵ ਕਿਉਂਕਿ ਆਰਜੀ ਤੌਰ ਤੇ ਬਣਾਇਆ ਗਿਆ ਟਾਸਕ ਸੀਕੁਇੰਗ ਕਦਮਾਂ ਵਿਚਕਾਰ ਜਾਣਕਾਰੀ ਨੂੰ ਪਾਸ ਕਰ ਸਕਦਾ ਹੈ. ਉਦਾਹਰਣ ਦੇ ਲਈ, ਇੱਕ ਈਟੀਐਲ ਪਾਈਪਲਾਈਨ ਵਿੱਚ, ਇੱਕ ਪ੍ਰੀਪ੍ਰੋਸੈਸਿੰਗ ਟਾਸਕ ਲੋੜੀਂਦਾ ਰੂਪਾਂਤਰਣ ਨਿਰਧਾਰਤ ਕਰ ਸਕਦਾ ਹੈ ਅਤੇ ਬਾਅਦ ਦੇ ਕੰਮਾਂ ਨੂੰ ਉਹਨਾਂ ਵੇਰਵਿਆਂ ਨੂੰ ਪਾਸ ਕਰ ਸਕਦਾ ਹੈ. ਇਹ ਵਿਧੀ ਸੱਚਮੁੱਚ ਡਾਟਾ-ਚਲਾਏ ਜਾ ਰਹੇ ਵਰਕਫਲੋਜ਼ ਨੂੰ ਸਮਰੱਥ ਬਣਾਉਂਦੀ ਹੈ, ਜਿੱਥੇ ਕਾਰਜਕਾਰੀ ਵਹਾਅ ਰੀਅਲ-ਟਾਈਮ ਇਨਪੁਟਸ ਦੇ ਅਧਾਰ ਤੇ, ਵਧਦੀ ਆਰਾਸ਼ਕਤੀ ਯੋਗਤਾ ਮਹੱਤਵਪੂਰਣ.

ਹਵਾ ਦੇ ਪ੍ਰਵਾਹ ਵਿੱਚ ਖੋਜ ਕਰਨ ਵਾਲੇ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਬਾਰੇ ਆਮ ਪ੍ਰਸ਼ਨ

ਕੀ ਹੈ dag_run.conf ਲਈ ਵਰਤਿਆ?
ਇਹ ਇੱਕ ਡੱਗ ਨੂੰ ਟਰਿੱਗਰ ਕਰਨ ਵੇਲੇ ਰਨਟਾਈਮ ਤੇ ਕੌਂਫਿਗਰੇਸ਼ਨ ਮਾਪਦੰਡ ਪਾਸ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ, ਵੂਸੀਫੋਜਾਂ ਨੂੰ ਵਧੇਰੇ ਲਚਕਦਾਰ ਬਣਾਉਂਦਾ ਹੈ.
ਮੈਂ ਆਰਜੀ ਤੌਰ ਤੇ ਏਅਰਫਲੋ ਵਿੱਚ ਕੰਮ ਕਿਵੇਂ ਪੈਦਾ ਕਰ ਸਕਦਾ ਹਾਂ?
ਤੁਸੀਂ ਏ ਦੇ ਕਈ ਉਦਾਹਰਣਾਂ ਨੂੰ ਤੁਰੰਤ ਕਰਨ ਲਈ ਇੱਕ ਲੂਪ ਦੀ ਵਰਤੋਂ ਕਰ ਸਕਦੇ ਹੋ PythonOperator ਜਾਂ ਵਰਤੋ @task ਟਾਸਕਫਲੋ ਏਪੀਆਈ ਵਿਚ ਸਜਾਵਟ.
ਵਰਤਣ ਦਾ ਕੀ ਲਾਭ ਹੈ BranchPythonOperator?
ਇਹ ਸ਼ਰਤੀਆ ਕੰਮਾਂ ਨੂੰ ਸਮਰੱਥ ਬਣਾਉਂਦਾ ਹੈ, ਤਾਂ ਡੀਗਾਂ ਨੂੰ ਪਰਿਭਾਸ਼ਿਤ ਤਰਕ, ਕੁਸ਼ਲਤਾ ਵਿੱਚ ਅਧਾਰਤ ਮਾਰਗਾਂ ਦੀ ਪਾਲਣਾ ਕਰਨ ਦੀ ਆਗਿਆ ਦੇਣਾ.
ਕਿਵੇਂ ਹੁੰਦਾ ਹੈ XComs ਗਤੀਸ਼ੀਲ ਡੈਗਸ ਵਧਾਉਣਾ?
Xcoms ਕਾਰਜਾਂ ਨੂੰ ਡੇਟਾ ਨੂੰ ਸਾਂਝਾ ਕਰਨ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ, ਇਹ ਸੁਨਿਸ਼ਚਿਤ ਕਰਦੇ ਹਨ ਕਿ ਪਿਛਲੇ ਪੜਾਅ ਤੋਂ ਸੰਬੰਧਿਤ ਜਾਣਕਾਰੀ ਪ੍ਰਾਪਤ ਕਰੋ.
ਕੀ ਮੈਂ ਨਿਰਭਰਤਾ ਨੂੰ ਆਰਜੀ ਤੌਰ ਤੇ ਨਿਰਧਾਰਤ ਕਰ ਸਕਦਾ ਹਾਂ?
ਹਾਂ, ਤੁਸੀਂ ਵਰਤ ਸਕਦੇ ਹੋ set_upstream() ਅਤੇ set_downstream() ਡੀਗ ਦੇ ਅੰਦਰ ਨਿਰਭਰਤਾ ਨੂੰ ਆਰਜੀ ਤੌਰ ਤੇ ਪਰਿਭਾਸ਼ਤ ਕਰਨ ਦੇ .ੰਗ.

ਰਨਟਾਈਮ ਕੌਂਫਿਗਰੇਸ਼ਨਾਂ ਨਾਲ ਗਤੀਸ਼ੀਲ ਵਰਕਫਲੋਅ ਨੂੰ ਅਨੁਕੂਲ ਬਣਾਉਣਾ

ਲਾਗੂ ਕਰਨਾ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਦਾ ਕੰਮ ਹਵਾ ਦੇ ਪ੍ਰਵਾਹ ਵਿੱਚ ਵਰਕਫਲੋ ਆਟੋਮੈਟੇਸ਼ਨ ਨੂੰ ਮਹੱਤਵਪੂਰਣ ਰੂਪ ਵਿੱਚ ਵਧਾਉਂਦਾ ਹੈ, ਉਹਨਾਂ ਨੂੰ ਤਬਦੀਲੀਆਂ ਨੂੰ ਬਦਲਣ ਲਈ ਅਨੁਕੂਲ ਬਣਾਉਂਦਾ ਹੈ. ਰੰਨਟਾਈਮ ਕੌਂਫਿਗਰੇਸ਼ਨਾਂ ਦਾ ਲਾਭ ਲੈ ਕੇ, ਡਿਵੈਲਪਰ ਸਟੈਟਿਕ ਡੀਗ ਦੀਆਂ ਪਰਿਭਾਸ਼ਾਵਾਂ ਤੋਂ ਪਰਹੇਜ਼ ਕਰ ਸਕਦੇ ਹਨ ਅਤੇ ਲਚਕਦਾਰ, ਡਾਟਾ-ਚਲਾਉਣ ਵਾਲੇ ਪਾਈਪ ਲਾਈਨਾਂ ਬਣਾ ਸਕਦੇ ਹਨ. ਇਹ ਪਹੁੰਚ ਖਾਸ ਤੌਰ 'ਤੇ ਵਾਤਾਵਰਣ ਵਿੱਚ ਮਹੱਤਵਪੂਰਣ ਹੈ ਜਿੱਥੇ ਕਾਰਜਾਂ ਨੂੰ ਰੀਅਲ-ਟਾਈਮ ਇਨਪੁਟ ਦੇ ਅਧਾਰ ਤੇ, ਵਿੱਤੀ ਰਿਪੋਰਟਿੰਗ ਜਾਂ ਮਸ਼ੀਨ ਸਿੱਖਣ ਦਾ ਮਾਡਲ ਸਿਖਲਾਈ ਦੇ ਅਧਾਰ ਤੇ ਪਰਿਭਾਸ਼ਤ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੁੰਦੀ ਹੈ. 🎯

ਏਕੀਕ੍ਰਿਤ ਕਰਕੇ dag_run.conf, ਕੰਡੀਸ਼ਨਲ ਫਾਂਸੀ, ਅਤੇ ਨਿਰਭਰਤਾ ਪ੍ਰਬੰਧਨ, ਟੀਮਾਂ ਸਕੇਲੇਬਲ ਅਤੇ ਕੁਸ਼ਲ ਵਰਕਫਲੋਸ ਬਣਾ ਸਕਦੀਆਂ ਹਨ. ਚਾਹੇ ਈ-ਕਾਮਰਸ ਟ੍ਰਾਂਜੈਕਸ਼ਨਾਂ 'ਤੇ ਪ੍ਰੋਸੈਸਿੰਗ ਕਰਦਿਆਂ, ਬੱਦਲ-ਅਧਾਰਤ ਡਾਟਾ ਟ੍ਰਾਂਸਫੋਰਸੇਸ਼ਨਜ਼, ਜਾਂ ਆਰਜ਼ੀਲ ਬੈਚ ਦੀਆਂ ਨੌਕਰੀਆਂ, ਏਅਰਫਲੋਅ ਦੀ ਗਤੀਸ਼ੀਲ ਡੱਗ ਸਮਰੱਥਾ ਪ੍ਰਦਾਨ ਕਰ ਰਹੇ ਹਨ. ਇਨ੍ਹਾਂ ਤਕਨੀਕਾਂ ਵਿਚ ਨਿਵੇਸ਼ ਕਰਨਾ ਕਾਰੋਬਾਰਾਂ ਨੂੰ ਮੈਨੂਅਲ ਦਖਲ ਘਟਾਉਣ ਦੌਰਾਨ ਕਾਰਜਾਂ ਨੂੰ ਸੁਚਾਰੂ ਬਣਾਉਣ ਦੀ ਆਗਿਆ ਦਿੰਦਾ ਹੈ.

ਏਅਰਫਲੋ ਵਿੱਚ ਸਰਮਾਲ ਵਿੱਚ ਕੰਮ ਕਰਨ ਵਾਲੇ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਲਈ ਸਰੋਤ ਅਤੇ ਹਵਾਲੇ

ਅਪਾਚੇ ਏਅਰਫਲੋ ਦਸਤਾਵੇਜ਼ਾਂ - DAG ਕੌਂਫਿਗਰੇਸ਼ਨ ਅਤੇ ਰਨਟਾਈਮ ਪੈਰਾਮੀਟਰਾਂ ਤੇ ਵਿਸਤ੍ਰਿਤ ਇਨਸਾਈਟਸ: ਅਪਾਚੇ ਏਅਰਫਲੋ ਅਧਿਕਾਰਤ ਡੌਕਸ
ਗਤੀਸ਼ੀਲ ਡੀਗ ਬਣਾਉਣ 'ਤੇ ਦਰਮਿਆਨੇ ਲੇਖ - ਵਰਤਣ' ਤੇ ਮਾਰਗਦਰਸ਼ਕ dag_run.conf ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਲਈ ਕੰਮ ਕਰਨ ਲਈ: ਮੀਡੀਅਮ: ਹਵਾ ਦੇ ਪ੍ਰਵਾਹ ਵਿੱਚ ਡਾਇਨਾਮਿਕ ਡੈਗਾਂ
ਇਨਪੁਟ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੇ ਅਧਾਰ ਤੇ ਆਰਜੀ ਤੌਰ ਤੇ ਤਿਆਰ ਕਰਨ ਲਈ ਅੰਤਰ-ਲਹਿਰਾਂ ਦੀ ਚਰਚਾ - ਆਰਜੀ ਤੌਰ ਤੇ ਤਿਆਰ ਕਰਨ ਲਈ ਕਮਿ Collections ਨਿਟੀ ਹੱਲ਼: ਸਟੈਕ ਓਵਰਫਲੋ ਧਾਗਾ
ਡਾਟਾ ਇੰਜੀਨੀਅਰਿੰਗ ਬਲਾੱਗ - ਸਕੇਲੇਬਲ ਏਅਰਫਲੋ ਵਰਕਫਲੋਜ਼ ਨੂੰ ਡਿਜ਼ਾਈਨ ਕਰਨ ਲਈ ਸਭ ਤੋਂ ਵਧੀਆ ਅਭਿਆਸ: ਡਾਟਾ ਇੰਜੀਨੀਅਰਿੰਗ ਬਲਾੱਗ

ਡੈਗ ਰਨ ਕੌਂਫਿਗਰੇਸ਼ਨ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਏਅਰਫਲੋ ਵਿੱਚ ਗਤੀਸ਼ੀਲ ਟਾਸਕ ਕ੍ਰਮ ਤਿਆਰ ਕਰਨਾ