🥇Apache Airflow: ETL ਨੂੰ ਆਸਾਨ ਬਣਾਉਣਾ

ਹੈਲੋ, ਮੈਂ ਦਮਿਤਰੀ ਲੌਗਵਿਨੇਨਕੋ ਹਾਂ - ਵੇਜ਼ੇਟ ਗਰੁੱਪ ਆਫ਼ ਕੰਪਨੀਆਂ ਦੇ ਵਿਸ਼ਲੇਸ਼ਣ ਵਿਭਾਗ ਦਾ ਡੇਟਾ ਇੰਜੀਨੀਅਰ।

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

ਅਤੇ ਹਾਂ, ਮੈਂ ਨਾ ਸਿਰਫ ਦੱਸਾਂਗਾ, ਬਲਕਿ ਇਹ ਵੀ ਦਿਖਾਵਾਂਗਾ: ਪ੍ਰੋਗਰਾਮ ਵਿੱਚ ਬਹੁਤ ਸਾਰੇ ਕੋਡ, ਸਕ੍ਰੀਨਸ਼ਾਟ ਅਤੇ ਸਿਫ਼ਾਰਸ਼ਾਂ ਹਨ.

ਜਦੋਂ ਤੁਸੀਂ ਏਅਰਫਲੋ / ਵਿਕੀਮੀਡੀਆ ਕਾਮਨਜ਼ ਸ਼ਬਦ ਨੂੰ ਗੂਗਲ ਕਰਦੇ ਹੋ ਤਾਂ ਤੁਸੀਂ ਆਮ ਤੌਰ 'ਤੇ ਕੀ ਦੇਖਦੇ ਹੋ

ਵਿਸ਼ਾ-ਸੂਚੀ

ਜਾਣ ਪਛਾਣ
ਮੁੱਖ ਹਿੱਸਾ, ਵਿਹਾਰਕ (ਅਤੇ ਥੋੜਾ ਸਿਧਾਂਤਕ)
ਭਾਗ ਫਾਈਨਲ, ਹਵਾਲਾ ਅਤੇ ਜਾਣਕਾਰੀ
- ਰੇਕ ਅਸੀਂ ਤੁਹਾਡੇ ਲਈ ਇਕੱਠਾ ਕੀਤਾ ਹੈ
- ਹੋਰ ਆਟੋਮੇਸ਼ਨ ਟੂਲ
ਹਵਾਲੇ

ਜਾਣ ਪਛਾਣ

ਅਪਾਚੇ ਏਅਰਫਲੋ ਜੈਂਗੋ ਵਾਂਗ ਹੈ:

python ਵਿੱਚ ਲਿਖਿਆ
ਇੱਕ ਵਧੀਆ ਐਡਮਿਨ ਪੈਨਲ ਹੈ,
ਅਨਿਸ਼ਚਿਤ ਸਮੇਂ ਲਈ ਵਿਸਤਾਰਯੋਗ

- ਸਿਰਫ ਬਿਹਤਰ, ਅਤੇ ਇਹ ਬਿਲਕੁਲ ਵੱਖਰੇ ਉਦੇਸ਼ਾਂ ਲਈ ਬਣਾਇਆ ਗਿਆ ਸੀ, ਅਰਥਾਤ (ਜਿਵੇਂ ਕਿ ਇਹ ਕਾਟਾ ਤੋਂ ਪਹਿਲਾਂ ਲਿਖਿਆ ਗਿਆ ਹੈ):

ਅਸੀਮਤ ਗਿਣਤੀ ਦੀਆਂ ਮਸ਼ੀਨਾਂ 'ਤੇ ਕੰਮ ਚਲਾਉਣਾ ਅਤੇ ਨਿਗਰਾਨੀ ਕਰਨਾ (ਜਿੰਨੇ ਸੈਲਰੀ / ਕੁਬਰਨੇਟਸ ਅਤੇ ਤੁਹਾਡੀ ਜ਼ਮੀਰ ਤੁਹਾਨੂੰ ਇਜਾਜ਼ਤ ਦੇਵੇਗੀ)
ਪਾਈਥਨ ਕੋਡ ਨੂੰ ਲਿਖਣ ਅਤੇ ਸਮਝਣ ਲਈ ਬਹੁਤ ਹੀ ਆਸਾਨ ਤੋਂ ਡਾਇਨਾਮਿਕ ਵਰਕਫਲੋ ਪੀੜ੍ਹੀ ਦੇ ਨਾਲ
ਅਤੇ ਕਿਸੇ ਵੀ ਡੇਟਾਬੇਸ ਅਤੇ API ਨੂੰ ਇੱਕ ਦੂਜੇ ਨਾਲ ਤਿਆਰ ਕੀਤੇ ਭਾਗਾਂ ਅਤੇ ਘਰੇਲੂ-ਬਣੇ ਪਲੱਗਇਨਾਂ (ਜੋ ਕਿ ਬਹੁਤ ਹੀ ਸਧਾਰਨ ਹੈ) ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਇੱਕ ਦੂਜੇ ਨਾਲ ਜੋੜਨ ਦੀ ਸਮਰੱਥਾ ਹੈ।

ਅਸੀਂ ਅਪਾਚੇ ਏਅਰਫਲੋ ਨੂੰ ਇਸ ਤਰ੍ਹਾਂ ਵਰਤਦੇ ਹਾਂ:

ਅਸੀਂ DWH ਅਤੇ ODS (ਸਾਡੇ ਕੋਲ Vertica ਅਤੇ Clickhouse ਹੈ) ਵਿੱਚ ਵੱਖ-ਵੱਖ ਸਰੋਤਾਂ (ਬਹੁਤ ਸਾਰੇ SQL ਸਰਵਰ ਅਤੇ PostgreSQL ਉਦਾਹਰਨਾਂ, ਐਪਲੀਕੇਸ਼ਨ ਮੈਟ੍ਰਿਕਸ ਦੇ ਨਾਲ ਵੱਖ-ਵੱਖ API, ਇੱਥੋਂ ਤੱਕ ਕਿ 1C) ਤੋਂ ਡਾਟਾ ਇਕੱਠਾ ਕਰਦੇ ਹਾਂ।
ਕਿੰਨਾ ਉੱਨਤ cron, ਜੋ ਕਿ ODS 'ਤੇ ਡਾਟਾ ਇਕਸੁਰਤਾ ਪ੍ਰਕਿਰਿਆਵਾਂ ਨੂੰ ਸ਼ੁਰੂ ਕਰਦਾ ਹੈ, ਅਤੇ ਉਹਨਾਂ ਦੇ ਰੱਖ-ਰਖਾਅ ਦੀ ਨਿਗਰਾਨੀ ਵੀ ਕਰਦਾ ਹੈ।

ਹਾਲ ਹੀ ਵਿੱਚ, ਸਾਡੀਆਂ ਲੋੜਾਂ ਨੂੰ 32 ਕੋਰ ਅਤੇ 50 GB RAM ਵਾਲੇ ਇੱਕ ਛੋਟੇ ਸਰਵਰ ਦੁਆਰਾ ਕਵਰ ਕੀਤਾ ਗਿਆ ਸੀ। ਏਅਰਫਲੋ ਵਿੱਚ, ਇਹ ਕੰਮ ਕਰਦਾ ਹੈ:

ਹੋਰ 200 ਡਾਗਸ (ਅਸਲ ਵਿੱਚ ਵਰਕਫਲੋ, ਜਿਸ ਵਿੱਚ ਅਸੀਂ ਕੰਮ ਭਰਦੇ ਹਾਂ),
ਔਸਤ 'ਤੇ ਹਰ ਵਿੱਚ 70 ਕਾਰਜ,
ਇਹ ਚੰਗਿਆਈ ਸ਼ੁਰੂ ਹੁੰਦੀ ਹੈ (ਔਸਤਨ ਵੀ) ਇੱਕ ਘੰਟੇ ਵਿੱਚ ਇੱਕ ਵਾਰ.

ਅਤੇ ਇਸ ਬਾਰੇ ਕਿ ਅਸੀਂ ਕਿਵੇਂ ਵਿਸਤਾਰ ਕੀਤਾ, ਮੈਂ ਹੇਠਾਂ ਲਿਖਾਂਗਾ, ਪਰ ਹੁਣ ਅਸੀਂ über-ਸਮੱਸਿਆ ਨੂੰ ਪਰਿਭਾਸ਼ਿਤ ਕਰੀਏ ਜਿਸ ਨੂੰ ਅਸੀਂ ਹੱਲ ਕਰਾਂਗੇ:

ਇੱਥੇ ਤਿੰਨ ਅਸਲੀ SQL ਸਰਵਰ ਹਨ, ਹਰ ਇੱਕ ਵਿੱਚ 50 ਡੇਟਾਬੇਸ ਹਨ - ਇੱਕ ਪ੍ਰੋਜੈਕਟ ਦੀਆਂ ਉਦਾਹਰਣਾਂ, ਕ੍ਰਮਵਾਰ, ਉਹਨਾਂ ਕੋਲ ਇੱਕੋ ਢਾਂਚਾ ਹੈ (ਲਗਭਗ ਹਰ ਥਾਂ, ਮੁਆ-ਹਾ-ਹਾ), ਜਿਸਦਾ ਮਤਲਬ ਹੈ ਕਿ ਹਰੇਕ ਕੋਲ ਇੱਕ ਆਰਡਰ ਟੇਬਲ ਹੈ (ਖੁਸ਼ਕਿਸਮਤੀ ਨਾਲ, ਉਸ ਨਾਲ ਇੱਕ ਸਾਰਣੀ) ਨਾਮ ਨੂੰ ਕਿਸੇ ਵੀ ਕਾਰੋਬਾਰ ਵਿੱਚ ਧੱਕਿਆ ਜਾ ਸਕਦਾ ਹੈ) ਅਸੀਂ ਸੇਵਾ ਖੇਤਰਾਂ (ਸਰੋਤ ਸਰਵਰ, ਸਰੋਤ ਡੇਟਾਬੇਸ, ETL ਟਾਸਕ ਆਈਡੀ) ਨੂੰ ਜੋੜ ਕੇ ਡੇਟਾ ਲੈਂਦੇ ਹਾਂ ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਨਿਰਪੱਖਤਾ ਨਾਲ ਵਰਟੀਕਾ ਵਿੱਚ ਸੁੱਟ ਦਿੰਦੇ ਹਾਂ।

ਚੱਲੀਏ!

ਮੁੱਖ ਹਿੱਸਾ, ਵਿਹਾਰਕ (ਅਤੇ ਥੋੜਾ ਸਿਧਾਂਤਕ)

ਅਸੀਂ (ਅਤੇ ਤੁਸੀਂ) ਕਿਉਂ ਕਰਦੇ ਹਾਂ

ਜਦੋਂ ਰੁੱਖ ਵੱਡੇ ਸਨ ਤੇ ਮੈਂ ਸਾਦਾ ਸੀ SQL-ਇੱਕ ਰੂਸੀ ਰਿਟੇਲ ਵਿੱਚ, ਅਸੀਂ ਸਾਡੇ ਲਈ ਉਪਲਬਧ ਦੋ ਟੂਲਸ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ETL ਪ੍ਰਕਿਰਿਆਵਾਂ ਉਰਫ਼ ਡਾਟਾ ਪ੍ਰਵਾਹ ਨਾਲ ਘਪਲਾ ਕੀਤਾ:

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

SQL ਸਰਵਰ ਏਕੀਕਰਣ ਸਰਵਰ - ਅਸੀਂ ਆਪਣੇ ਇੰਟਰਾ-ਪ੍ਰੋਜੈਕਟ ਪ੍ਰਵਾਹ ਵਿੱਚ ਇਸ ਕਾਮਰੇਡ ਦੀ ਵਰਤੋਂ ਕੀਤੀ। ਖੈਰ, ਵਾਸਤਵ ਵਿੱਚ: ਅਸੀਂ ਪਹਿਲਾਂ ਹੀ SQL ਸਰਵਰ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹਾਂ, ਅਤੇ ਇਸਦੇ ETL ਸਾਧਨਾਂ ਦੀ ਵਰਤੋਂ ਨਾ ਕਰਨਾ ਕਿਸੇ ਤਰ੍ਹਾਂ ਗੈਰਵਾਜਬ ਹੋਵੇਗਾ. ਇਸ ਵਿੱਚ ਸਭ ਕੁਝ ਵਧੀਆ ਹੈ: ਦੋਵੇਂ ਇੰਟਰਫੇਸ ਸੁੰਦਰ ਹਨ, ਅਤੇ ਪ੍ਰਗਤੀ ਰਿਪੋਰਟਾਂ ... ਪਰ ਇਹ ਇਸ ਲਈ ਨਹੀਂ ਹੈ ਕਿ ਅਸੀਂ ਸਾਫਟਵੇਅਰ ਉਤਪਾਦਾਂ ਨੂੰ ਪਿਆਰ ਕਰਦੇ ਹਾਂ, ਓ, ਇਸਦੇ ਲਈ ਨਹੀਂ. ਇਸਦਾ ਸੰਸਕਰਣ dtsx (ਜੋ ਕਿ ਨੋਡਸ ਨੂੰ ਸੇਵ ਕਰਨ 'ਤੇ ਬਦਲਿਆ ਹੋਇਆ XML ਹੈ) ਅਸੀਂ ਕਰ ਸਕਦੇ ਹਾਂ, ਪਰ ਇਸ ਦਾ ਕੀ ਮਤਲਬ ਹੈ? ਇੱਕ ਟਾਸਕ ਪੈਕੇਜ ਬਣਾਉਣ ਬਾਰੇ ਕੀ ਹੈ ਜੋ ਇੱਕ ਸਰਵਰ ਤੋਂ ਦੂਜੇ ਸਰਵਰ ਵਿੱਚ ਸੈਂਕੜੇ ਟੇਬਲਾਂ ਨੂੰ ਖਿੱਚੇਗਾ? ਹਾਂ, ਕੀ ਸੌ, ਤੁਹਾਡੀ ਇੰਡੈਕਸ ਫਿੰਗਰ ਵੀਹ ਟੁਕੜਿਆਂ ਤੋਂ ਡਿੱਗ ਜਾਵੇਗੀ, ਮਾਊਸ ਬਟਨ 'ਤੇ ਕਲਿੱਕ ਕਰੋ. ਪਰ ਇਹ ਯਕੀਨੀ ਤੌਰ 'ਤੇ ਵਧੇਰੇ ਫੈਸ਼ਨੇਬਲ ਦਿਖਾਈ ਦਿੰਦਾ ਹੈ:

ਅਸੀਂ ਨਿਸ਼ਚਤ ਤੌਰ 'ਤੇ ਬਾਹਰ ਦੇ ਤਰੀਕਿਆਂ ਦੀ ਭਾਲ ਕੀਤੀ. ਕੇਸ ਵੀ ਲਗਭਗ ਲਗਭਗ ਇੱਕ ਸਵੈ-ਲਿਖਤ SSIS ਪੈਕੇਜ ਜਨਰੇਟਰ ਤੇ ਆਇਆ ...

…ਅਤੇ ਫਿਰ ਮੈਨੂੰ ਇੱਕ ਨਵੀਂ ਨੌਕਰੀ ਮਿਲੀ। ਅਤੇ ਅਪਾਚੇ ਏਅਰਫਲੋ ਨੇ ਮੈਨੂੰ ਇਸ 'ਤੇ ਪਛਾੜ ਦਿੱਤਾ.

ਜਦੋਂ ਮੈਨੂੰ ਪਤਾ ਲੱਗਾ ਕਿ ETL ਪ੍ਰਕਿਰਿਆ ਦੇ ਵਰਣਨ ਸਧਾਰਨ ਪਾਈਥਨ ਕੋਡ ਹਨ, ਤਾਂ ਮੈਂ ਖੁਸ਼ੀ ਲਈ ਨੱਚਿਆ ਨਹੀਂ ਸੀ. ਇਸ ਤਰ੍ਹਾਂ ਡੇਟਾ ਸਟ੍ਰੀਮਾਂ ਦਾ ਸੰਸਕਰਣ ਅਤੇ ਵੱਖਰਾ ਕੀਤਾ ਗਿਆ ਸੀ, ਅਤੇ ਸੈਂਕੜੇ ਡੇਟਾਬੇਸ ਤੋਂ ਇੱਕ ਸਿੰਗਲ ਢਾਂਚੇ ਦੇ ਨਾਲ ਟੇਬਲ ਨੂੰ ਇੱਕ ਟੀਚੇ ਵਿੱਚ ਪਾਉਣਾ ਡੇਢ ਜਾਂ ਦੋ 13 ”ਸਕ੍ਰੀਨਾਂ ਵਿੱਚ ਪਾਈਥਨ ਕੋਡ ਦਾ ਮਾਮਲਾ ਬਣ ਗਿਆ ਸੀ।

ਕਲੱਸਟਰ ਨੂੰ ਇਕੱਠਾ ਕਰਨਾ

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

ਤਾਂ ਜੋ ਅਸੀਂ ਤੁਰੰਤ ਪ੍ਰਯੋਗ ਸ਼ੁਰੂ ਕਰ ਸਕੀਏ, ਮੈਂ ਸਕੈਚ ਕੀਤਾ docker-compose.yml ਜਿਸ ਵਿੱਚ:

ਦੇ ਅਸਲ ਵਿੱਚ ਉਠਾਉਣ ਕਰੀਏ airflow: ਸ਼ਡਿਊਲਰ, ਵੈਬਸਰਵਰ। ਸੈਲਰੀ ਦੇ ਕੰਮਾਂ ਦੀ ਨਿਗਰਾਨੀ ਕਰਨ ਲਈ ਫਲਾਵਰ ਵੀ ਉੱਥੇ ਘੁੰਮ ਰਿਹਾ ਹੋਵੇਗਾ (ਕਿਉਂਕਿ ਇਹ ਪਹਿਲਾਂ ਹੀ ਇਸ ਵਿੱਚ ਧੱਕਿਆ ਜਾ ਚੁੱਕਾ ਹੈ apache/airflow:1.10.10-python3.7, ਪਰ ਸਾਨੂੰ ਕੋਈ ਇਤਰਾਜ਼ ਨਹੀਂ ਹੈ)
PostgreSQL, ਜਿਸ ਵਿੱਚ ਏਅਰਫਲੋ ਆਪਣੀ ਸੇਵਾ ਜਾਣਕਾਰੀ (ਸ਼ਡਿਊਲਰ ਡੇਟਾ, ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਸਟੈਟਿਸਟਿਕਸ, ਆਦਿ) ਲਿਖੇਗਾ, ਅਤੇ ਸੈਲਰੀ ਪੂਰੇ ਕੀਤੇ ਕੰਮਾਂ ਨੂੰ ਚਿੰਨ੍ਹਿਤ ਕਰੇਗੀ;
ਰੇਡਿਸ, ਜੋ ਸੈਲਰੀ ਲਈ ਟਾਸਕ ਬ੍ਰੋਕਰ ਵਜੋਂ ਕੰਮ ਕਰੇਗਾ;
ਸੈਲਰੀ ਵਰਕਰ, ਜੋ ਕਿ ਕੰਮਾਂ ਦੇ ਸਿੱਧੇ ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਵਿੱਚ ਲੱਗੇਗਾ।
ਫੋਲਡਰ ਨੂੰ ./dags ਅਸੀਂ ਆਪਣੀਆਂ ਫਾਈਲਾਂ ਨੂੰ ਡੈਗਸ ਦੇ ਵਰਣਨ ਨਾਲ ਜੋੜਾਂਗੇ। ਉਹਨਾਂ ਨੂੰ ਉੱਡਣ 'ਤੇ ਚੁੱਕਿਆ ਜਾਵੇਗਾ, ਇਸ ਲਈ ਹਰ ਛਿੱਕ ਦੇ ਬਾਅਦ ਪੂਰੇ ਸਟੈਕ ਨੂੰ ਜੁਗਲ ਕਰਨ ਦੀ ਕੋਈ ਲੋੜ ਨਹੀਂ ਹੈ।

ਕੁਝ ਥਾਵਾਂ 'ਤੇ, ਉਦਾਹਰਨਾਂ ਵਿੱਚ ਕੋਡ ਪੂਰੀ ਤਰ੍ਹਾਂ ਨਹੀਂ ਦਿਖਾਇਆ ਗਿਆ ਹੈ (ਤਾਂ ਕਿ ਟੈਕਸਟ ਵਿੱਚ ਗੜਬੜ ਨਾ ਹੋਵੇ), ਪਰ ਕਿਤੇ ਇਹ ਪ੍ਰਕਿਰਿਆ ਵਿੱਚ ਸੋਧਿਆ ਗਿਆ ਹੈ। ਸੰਪੂਰਨ ਕਾਰਜ ਕੋਡ ਉਦਾਹਰਨਾਂ ਰਿਪੋਜ਼ਟਰੀ ਵਿੱਚ ਲੱਭੀਆਂ ਜਾ ਸਕਦੀਆਂ ਹਨ https://github.com/dm-logv/airflow-tutorial.

ਡੌਕਰ-ਕੰਪੋਜ਼.ਆਈ.ਐਮ.ਐਲ.

version: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

services:
  # Redis as a Celery broker
  broker:
    image: redis:6.0.5-alpine

  # DB for the Airflow metadata
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Main container with Airflow Webserver, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <<: *airflow-config

      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'
      AIRFLOW__SCHEDULER__MAX_THREADS: 8

      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10

    depends_on:
      - airflow-db
      - broker

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Airflow Webserver
      - 8080:8080

  # Celery worker, will be scaled using `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <<: *airflow-config

    command: >
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

ਟਿੱਪਣੀ:

ਰਚਨਾ ਦੀ ਅਸੈਂਬਲੀ ਵਿੱਚ, ਮੈਂ ਕਾਫ਼ੀ ਹੱਦ ਤੱਕ ਜਾਣੇ-ਪਛਾਣੇ ਚਿੱਤਰ 'ਤੇ ਭਰੋਸਾ ਕੀਤਾ puckel/docker- airflow - ਇਸਦੀ ਜਾਂਚ ਕਰਨਾ ਯਕੀਨੀ ਬਣਾਓ। ਹੋ ਸਕਦਾ ਹੈ ਕਿ ਤੁਹਾਨੂੰ ਆਪਣੀ ਜ਼ਿੰਦਗੀ ਵਿੱਚ ਹੋਰ ਕਿਸੇ ਚੀਜ਼ ਦੀ ਲੋੜ ਨਾ ਹੋਵੇ।
ਸਾਰੀਆਂ ਏਅਰਫਲੋ ਸੈਟਿੰਗਾਂ ਨਾ ਸਿਰਫ਼ ਰਾਹੀਂ ਉਪਲਬਧ ਹਨ airflow.cfg, ਪਰ ਵਾਤਾਵਰਣ ਵੇਰੀਏਬਲ (ਡਿਵੈਲਪਰਾਂ ਦਾ ਧੰਨਵਾਦ) ਦੁਆਰਾ ਵੀ, ਜਿਸਦਾ ਮੈਂ ਗਲਤ ਤਰੀਕੇ ਨਾਲ ਫਾਇਦਾ ਉਠਾਇਆ।
ਕੁਦਰਤੀ ਤੌਰ 'ਤੇ, ਇਹ ਉਤਪਾਦਨ ਲਈ ਤਿਆਰ ਨਹੀਂ ਹੈ: ਮੈਂ ਜਾਣਬੁੱਝ ਕੇ ਕੰਟੇਨਰਾਂ 'ਤੇ ਦਿਲ ਦੀ ਧੜਕਣ ਨਹੀਂ ਪਾਈ, ਮੈਂ ਸੁਰੱਖਿਆ ਨਾਲ ਪਰੇਸ਼ਾਨ ਨਹੀਂ ਕੀਤਾ. ਪਰ ਮੈਂ ਸਾਡੇ ਪ੍ਰਯੋਗ ਕਰਨ ਵਾਲਿਆਂ ਲਈ ਘੱਟੋ-ਘੱਟ ਢੁਕਵਾਂ ਕੀਤਾ।
ਨੋਟ ਕਰੋ:
- ਡੈਗ ਫੋਲਡਰ ਸ਼ਡਿਊਲਰ ਅਤੇ ਵਰਕਰਾਂ ਦੋਵਾਂ ਲਈ ਪਹੁੰਚਯੋਗ ਹੋਣਾ ਚਾਹੀਦਾ ਹੈ।
- ਇਹੀ ਸਾਰੀਆਂ ਤੀਜੀ-ਧਿਰ ਲਾਇਬ੍ਰੇਰੀਆਂ 'ਤੇ ਲਾਗੂ ਹੁੰਦਾ ਹੈ - ਉਹ ਸਾਰੀਆਂ ਮਸ਼ੀਨਾਂ 'ਤੇ ਇੱਕ ਸ਼ਡਿਊਲਰ ਅਤੇ ਵਰਕਰਾਂ ਨਾਲ ਸਥਾਪਿਤ ਹੋਣੀਆਂ ਚਾਹੀਦੀਆਂ ਹਨ।

ਖੈਰ, ਹੁਣ ਇਹ ਸਧਾਰਨ ਹੈ:

$ docker-compose up --scale worker=3

ਸਭ ਕੁਝ ਵਧਣ ਤੋਂ ਬਾਅਦ, ਤੁਸੀਂ ਵੈੱਬ ਇੰਟਰਫੇਸਾਂ ਨੂੰ ਦੇਖ ਸਕਦੇ ਹੋ:

ਹਵਾ ਦਾ ਪ੍ਰਵਾਹ: http://127.0.0.1:8080/admin/
ਫੁੱਲ: http://127.0.0.1:5555/dashboard

ਬੇਸਿਕ ਧਾਰਨਾ

ਜੇ ਤੁਸੀਂ ਇਹਨਾਂ ਸਾਰੇ "ਡੈਗਜ਼" ਵਿੱਚ ਕੁਝ ਨਹੀਂ ਸਮਝਿਆ, ਤਾਂ ਇੱਥੇ ਇੱਕ ਛੋਟਾ ਸ਼ਬਦਕੋਸ਼ ਹੈ:

ਸੈਡਿਊਲਰ - ਏਅਰਫਲੋ ਵਿੱਚ ਸਭ ਤੋਂ ਮਹੱਤਵਪੂਰਨ ਅੰਕਲ, ਇਹ ਨਿਯੰਤਰਿਤ ਕਰਨਾ ਕਿ ਰੋਬੋਟ ਸਖਤ ਮਿਹਨਤ ਕਰਦੇ ਹਨ, ਨਾ ਕਿ ਇੱਕ ਵਿਅਕਤੀ: ਸਮਾਂ-ਸੂਚੀ ਦੀ ਨਿਗਰਾਨੀ ਕਰਦਾ ਹੈ, ਡੈਗਜ਼ ਨੂੰ ਅਪਡੇਟ ਕਰਦਾ ਹੈ, ਕੰਮ ਸ਼ੁਰੂ ਕਰਦਾ ਹੈ।
ਆਮ ਤੌਰ 'ਤੇ, ਪੁਰਾਣੇ ਸੰਸਕਰਣਾਂ ਵਿੱਚ, ਉਸਨੂੰ ਯਾਦਦਾਸ਼ਤ ਨਾਲ ਸਮੱਸਿਆਵਾਂ ਸਨ (ਨਹੀਂ, ਐਮਨੇਸ਼ੀਆ ਨਹੀਂ, ਪਰ ਲੀਕ) ਅਤੇ ਵਿਰਾਸਤੀ ਪੈਰਾਮੀਟਰ ਸੰਰਚਨਾ ਵਿੱਚ ਵੀ ਰਿਹਾ। run_duration - ਇਸਦਾ ਮੁੜ-ਚਾਲੂ ਅੰਤਰਾਲ। ਪਰ ਹੁਣ ਸਭ ਕੁਝ ਠੀਕ ਹੈ।
ਡੀ.ਏ.ਜੀ. (ਉਰਫ਼ "ਡੈਗ") - "ਡਾਇਰੈਕਟਡ ਐਸੀਕਲਿਕ ਗ੍ਰਾਫ", ਪਰ ਅਜਿਹੀ ਪਰਿਭਾਸ਼ਾ ਕੁਝ ਲੋਕਾਂ ਨੂੰ ਦੱਸੇਗੀ, ਪਰ ਅਸਲ ਵਿੱਚ ਇਹ ਇੱਕ ਦੂਜੇ ਨਾਲ ਇੰਟਰੈਕਟ ਕਰਨ ਵਾਲੇ ਕੰਮਾਂ ਲਈ ਇੱਕ ਕੰਟੇਨਰ ਹੈ (ਹੇਠਾਂ ਦੇਖੋ) ਜਾਂ SSIS ਵਿੱਚ ਪੈਕੇਜ ਦਾ ਐਨਾਲਾਗ ਅਤੇ ਇਨਫਾਰਮੈਟਿਕਾ ਵਿੱਚ ਵਰਕਫਲੋ .
ਡੈਗਸ ਤੋਂ ਇਲਾਵਾ, ਅਜੇ ਵੀ ਸਬਡੈਗ ਹੋ ਸਕਦੇ ਹਨ, ਪਰ ਅਸੀਂ ਸੰਭਾਵਤ ਤੌਰ 'ਤੇ ਉਨ੍ਹਾਂ ਤੱਕ ਨਹੀਂ ਪਹੁੰਚਾਂਗੇ।
ਡੀਏਜੀ ਰਨ - ਸ਼ੁਰੂਆਤੀ ਡੈਗ, ਜਿਸ ਨੂੰ ਆਪਣਾ ਨਿਰਧਾਰਤ ਕੀਤਾ ਗਿਆ ਹੈ execution_date. ਇੱਕੋ ਡੈਗ ਦੇ ਡੈਗਰਨ ਸਮਾਨਾਂਤਰ ਵਿੱਚ ਕੰਮ ਕਰ ਸਕਦੇ ਹਨ (ਜੇ ਤੁਸੀਂ ਆਪਣੇ ਕਾਰਜਾਂ ਨੂੰ ਬੇਸ਼ਕ, ਬੇਸ਼ਕ) ਬਣਾਇਆ ਹੈ।
ਓਪਰੇਟਰ ਇੱਕ ਖਾਸ ਕਾਰਵਾਈ ਕਰਨ ਲਈ ਜ਼ਿੰਮੇਵਾਰ ਕੋਡ ਦੇ ਟੁਕੜੇ ਹਨ। ਓਪਰੇਟਰਾਂ ਦੀਆਂ ਤਿੰਨ ਕਿਸਮਾਂ ਹਨ:
- ਕਾਰਵਾਈਸਾਡੇ ਮਨਪਸੰਦ ਵਾਂਗ PythonOperator, ਜੋ ਕਿਸੇ ਵੀ (ਵੈਧ) ਪਾਈਥਨ ਕੋਡ ਨੂੰ ਚਲਾ ਸਕਦਾ ਹੈ;
- ਟ੍ਰਾਂਸਫਰ, ਜੋ ਡੇਟਾ ਨੂੰ ਥਾਂ-ਥਾਂ ਟ੍ਰਾਂਸਪੋਰਟ ਕਰਦੇ ਹਨ, ਕਹਿੰਦੇ ਹਨ, MsSqlToHiveTransfer;
- ਸੂਚਕ ਦੂਜੇ ਪਾਸੇ, ਇਹ ਤੁਹਾਨੂੰ ਕੋਈ ਘਟਨਾ ਵਾਪਰਨ ਤੱਕ ਡੈਗ ਦੇ ਅਗਲੇ ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਨੂੰ ਪ੍ਰਤੀਕਿਰਿਆ ਕਰਨ ਜਾਂ ਹੌਲੀ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦੇਵੇਗਾ। HttpSensor ਨਿਰਧਾਰਤ ਅੰਤਮ ਬਿੰਦੂ ਨੂੰ ਖਿੱਚ ਸਕਦਾ ਹੈ, ਅਤੇ ਜਦੋਂ ਲੋੜੀਂਦਾ ਜਵਾਬ ਉਡੀਕ ਕਰ ਰਿਹਾ ਹੋਵੇ, ਟ੍ਰਾਂਸਫਰ ਸ਼ੁਰੂ ਕਰੋ GoogleCloudStorageToS3Operator. ਇੱਕ ਖੋਜੀ ਮਨ ਪੁੱਛੇਗਾ: “ਕਿਉਂ? ਆਖਰਕਾਰ, ਤੁਸੀਂ ਆਪਰੇਟਰ ਵਿੱਚ ਹੀ ਦੁਹਰਾਓ ਕਰ ਸਕਦੇ ਹੋ!” ਅਤੇ ਫਿਰ, ਮੁਅੱਤਲ ਕੀਤੇ ਓਪਰੇਟਰਾਂ ਦੇ ਨਾਲ ਕਾਰਜਾਂ ਦੇ ਪੂਲ ਨੂੰ ਬੰਦ ਨਾ ਕਰਨ ਲਈ. ਅਗਲੀ ਕੋਸ਼ਿਸ਼ ਤੋਂ ਪਹਿਲਾਂ ਸੈਂਸਰ ਸ਼ੁਰੂ ਹੁੰਦਾ ਹੈ, ਜਾਂਚ ਕਰਦਾ ਹੈ ਅਤੇ ਮਰ ਜਾਂਦਾ ਹੈ।
ਟਾਸਕ - ਘੋਸ਼ਿਤ ਓਪਰੇਟਰ, ਕਿਸਮ ਦੀ ਪਰਵਾਹ ਕੀਤੇ ਬਿਨਾਂ, ਅਤੇ ਡੈਗ ਨਾਲ ਜੁੜੇ ਹੋਏ ਨੂੰ ਕਾਰਜ ਦੇ ਦਰਜੇ 'ਤੇ ਤਰੱਕੀ ਦਿੱਤੀ ਜਾਂਦੀ ਹੈ।
ਕਾਰਜ ਉਦਾਹਰਨ - ਜਦੋਂ ਆਮ ਯੋਜਨਾਕਾਰ ਨੇ ਫੈਸਲਾ ਕੀਤਾ ਕਿ ਇਹ ਕੰਮ ਕਰਨ ਵਾਲੇ-ਕਰਮਚਾਰੀਆਂ ਨੂੰ ਲੜਾਈ ਵਿੱਚ ਭੇਜਣ ਦਾ ਸਮਾਂ ਹੈ (ਸਹੀ ਮੌਕੇ 'ਤੇ, ਜੇ ਅਸੀਂ ਵਰਤਦੇ ਹਾਂ LocalExecutor ਜਾਂ ਦੇ ਮਾਮਲੇ ਵਿੱਚ ਇੱਕ ਰਿਮੋਟ ਨੋਡ ਤੱਕ CeleryExecutor), ਇਹ ਉਹਨਾਂ ਨੂੰ ਇੱਕ ਸੰਦਰਭ ਨਿਰਧਾਰਤ ਕਰਦਾ ਹੈ (ਅਰਥਾਤ, ਵੇਰੀਏਬਲਾਂ ਦਾ ਇੱਕ ਸੈੱਟ - ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਪੈਰਾਮੀਟਰ), ਕਮਾਂਡ ਜਾਂ ਪੁੱਛਗਿੱਛ ਟੈਂਪਲੇਟਸ ਦਾ ਵਿਸਤਾਰ ਕਰਦਾ ਹੈ, ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਪੂਲ ਕਰਦਾ ਹੈ।

ਅਸੀਂ ਕਾਰਜ ਤਿਆਰ ਕਰਦੇ ਹਾਂ

ਪਹਿਲਾਂ, ਆਓ ਆਪਣੇ ਡੌਗ ਦੀ ਆਮ ਸਕੀਮ ਦੀ ਰੂਪਰੇਖਾ ਕਰੀਏ, ਅਤੇ ਫਿਰ ਅਸੀਂ ਵਧੇਰੇ ਅਤੇ ਹੋਰ ਵੇਰਵਿਆਂ ਵਿੱਚ ਡੁਬਕੀ ਲਵਾਂਗੇ, ਕਿਉਂਕਿ ਅਸੀਂ ਕੁਝ ਗੈਰ-ਮਾਮੂਲੀ ਹੱਲ ਲਾਗੂ ਕਰਦੇ ਹਾਂ.

ਇਸ ਲਈ, ਇਸਦੇ ਸਰਲ ਰੂਪ ਵਿੱਚ, ਅਜਿਹਾ ਡੈਗ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦੇਵੇਗਾ:

from datetime import timedelta, datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

from commons.datasources import sql_server_ds

dag = DAG('orders',
          schedule_interval=timedelta(hours=6),
          start_date=datetime(2020, 7, 8, 0))

def workflow(**context):
    print(context)

for conn_id, schema in sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

ਆਓ ਇਸਦਾ ਪਤਾ ਕਰੀਏ:

ਪਹਿਲਾਂ, ਅਸੀਂ ਲੋੜੀਂਦੇ libs ਅਤੇ ਆਯਾਤ ਕਰਦੇ ਹਾਂ ਕੁਝ ਹੋਰ;
sql_server_ds ਹੈ List[namedtuple[str, str]] ਏਅਰਫਲੋ ਕਨੈਕਸ਼ਨਾਂ ਦੇ ਕਨੈਕਸ਼ਨਾਂ ਦੇ ਨਾਵਾਂ ਅਤੇ ਡੇਟਾਬੇਸ ਦੇ ਨਾਲ ਜਿਨ੍ਹਾਂ ਤੋਂ ਅਸੀਂ ਆਪਣੀ ਪਲੇਟ ਲਵਾਂਗੇ;
dag - ਸਾਡੇ ਡੇਗ ਦੀ ਘੋਸ਼ਣਾ, ਜੋ ਜ਼ਰੂਰੀ ਤੌਰ 'ਤੇ ਹੋਣੀ ਚਾਹੀਦੀ ਹੈ globals(), ਨਹੀਂ ਤਾਂ ਏਅਰਫਲੋ ਇਸ ਨੂੰ ਨਹੀਂ ਲੱਭ ਸਕੇਗਾ। ਡੱਗ ਨੂੰ ਇਹ ਵੀ ਕਹਿਣ ਦੀ ਲੋੜ ਹੈ:
- ਉਸਦਾ ਨਾਮ ਕੀ ਹੈ orders - ਇਹ ਨਾਮ ਫਿਰ ਵੈੱਬ ਇੰਟਰਫੇਸ ਵਿੱਚ ਦਿਖਾਈ ਦੇਵੇਗਾ,
- ਕਿ ਉਹ ਅੱਠ ਜੁਲਾਈ ਦੀ ਅੱਧੀ ਰਾਤ ਤੋਂ ਕੰਮ ਕਰੇਗਾ,
- ਅਤੇ ਇਹ ਲਗਭਗ ਹਰ 6 ਘੰਟਿਆਂ ਬਾਅਦ ਚੱਲਣਾ ਚਾਹੀਦਾ ਹੈ (ਇਸ ਦੀ ਬਜਾਏ ਇੱਥੇ ਸਖ਼ਤ ਮੁੰਡਿਆਂ ਲਈ timedelta() ਇਜਾਜ਼ਤ ਹੈ cron-ਲਾਈਨ 0 0 0/6 ? * * *, ਘੱਟ ਠੰਡੇ ਲਈ - ਇੱਕ ਸਮੀਕਰਨ ਵਰਗਾ @daily);
workflow() ਮੁੱਖ ਕੰਮ ਕਰੇਗਾ, ਪਰ ਹੁਣ ਨਹੀਂ। ਹੁਣ ਲਈ, ਅਸੀਂ ਆਪਣੇ ਸੰਦਰਭ ਨੂੰ ਲੌਗ ਵਿੱਚ ਡੰਪ ਕਰਾਂਗੇ।
ਅਤੇ ਹੁਣ ਕਾਰਜ ਬਣਾਉਣ ਦਾ ਸਧਾਰਨ ਜਾਦੂ:
- ਅਸੀਂ ਆਪਣੇ ਸਰੋਤਾਂ ਰਾਹੀਂ ਚੱਲਦੇ ਹਾਂ;
- ਸ਼ੁਰੂਆਤ PythonOperator, ਜੋ ਸਾਡੇ ਡਮੀ ਨੂੰ ਚਲਾਏਗਾ workflow(). ਕੰਮ ਦਾ ਇੱਕ ਵਿਲੱਖਣ (ਡੈਗ ਦੇ ਅੰਦਰ) ਨਾਮ ਨਿਰਧਾਰਤ ਕਰਨਾ ਨਾ ਭੁੱਲੋ ਅਤੇ ਡੈਗ ਨੂੰ ਆਪਣੇ ਆਪ ਬੰਨ੍ਹੋ। ਝੰਡਾ provide_context ਬਦਲੇ ਵਿੱਚ, ਫੰਕਸ਼ਨ ਵਿੱਚ ਵਾਧੂ ਆਰਗੂਮੈਂਟ ਪਾਵੇਗਾ, ਜਿਸਨੂੰ ਅਸੀਂ ਧਿਆਨ ਨਾਲ ਵਰਤ ਕੇ ਇਕੱਠਾ ਕਰਾਂਗੇ **context.

ਹੁਣ ਲਈ, ਇਹ ਸਭ ਹੈ. ਸਾਨੂੰ ਕੀ ਮਿਲਿਆ:

ਵੈੱਬ ਇੰਟਰਫੇਸ ਵਿੱਚ ਨਵਾਂ ਡੈਗ,
ਡੇਢ ਸੌ ਕਾਰਜ ਜੋ ਸਮਾਨਾਂਤਰ ਤੌਰ 'ਤੇ ਲਾਗੂ ਕੀਤੇ ਜਾਣਗੇ (ਜੇ ਏਅਰਫਲੋ, ਸੈਲਰੀ ਸੈਟਿੰਗਾਂ ਅਤੇ ਸਰਵਰ ਸਮਰੱਥਾ ਇਸਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦੇ ਹਨ)।

ਖੈਰ, ਲਗਭਗ ਇਸ ਨੂੰ ਮਿਲ ਗਿਆ.

ਨਿਰਭਰਤਾਵਾਂ ਨੂੰ ਕੌਣ ਸਥਾਪਿਤ ਕਰੇਗਾ?

ਇਸ ਸਾਰੀ ਗੱਲ ਨੂੰ ਸਰਲ ਬਣਾਉਣ ਲਈ, ਮੈਂ ਅੰਦਰ ਘੁਸਪੈਠ ਕੀਤੀ docker-compose.yml ਕਾਰਵਾਈ requirements.txt ਸਾਰੇ ਨੋਡਾਂ 'ਤੇ.

ਹੁਣ ਇਹ ਚਲਾ ਗਿਆ ਹੈ:

ਸਲੇਟੀ ਵਰਗ ਸ਼ਡਿਊਲਰ ਦੁਆਰਾ ਸੰਸਾਧਿਤ ਕੀਤੇ ਗਏ ਟਾਸਕ ਉਦਾਹਰਨ ਹਨ।

ਅਸੀਂ ਥੋੜਾ ਇੰਤਜ਼ਾਰ ਕਰਦੇ ਹਾਂ, ਵਰਕਰਾਂ ਦੁਆਰਾ ਕੰਮ ਬੰਦ ਕਰ ਦਿੱਤੇ ਜਾਂਦੇ ਹਨ:

ਹਰੇ ਵਾਲੇ, ਬੇਸ਼ੱਕ, ਸਫਲਤਾਪੂਰਵਕ ਆਪਣਾ ਕੰਮ ਪੂਰਾ ਕਰ ਚੁੱਕੇ ਹਨ. ਲਾਲ ਬਹੁਤ ਸਫਲ ਨਹੀਂ ਹਨ.

ਤਰੀਕੇ ਨਾਲ, ਸਾਡੇ ਉਤਪਾਦ 'ਤੇ ਕੋਈ ਫੋਲਡਰ ਨਹੀਂ ਹੈ ./dags, ਮਸ਼ੀਨਾਂ ਵਿਚਕਾਰ ਕੋਈ ਸਮਕਾਲੀਕਰਨ ਨਹੀਂ ਹੈ - ਸਾਰੇ ਡਾਗ ਅੰਦਰ ਪਏ ਹਨ git ਸਾਡੀ ਗਿਟਲੈਬ 'ਤੇ, ਅਤੇ ਗਿਟਲੈਬ ਸੀਆਈ ਮਸ਼ੀਨਾਂ ਨੂੰ ਅੱਪਡੇਟ ਵੰਡਦਾ ਹੈ ਜਦੋਂ ਇਸ ਵਿੱਚ ਅਭੇਦ ਹੁੰਦਾ ਹੈ master.

ਫੁੱਲ ਬਾਰੇ ਥੋੜਾ ਜਿਹਾ

ਜਦੋਂ ਕਿ ਵਰਕਰ ਸਾਡੇ ਸ਼ਾਂਤ ਕਰਨ ਵਾਲਿਆਂ ਨੂੰ ਕੁੱਟ ਰਹੇ ਹਨ, ਆਓ ਇਕ ਹੋਰ ਸਾਧਨ ਨੂੰ ਯਾਦ ਕਰੀਏ ਜੋ ਸਾਨੂੰ ਕੁਝ ਦਿਖਾ ਸਕਦਾ ਹੈ - ਫਲਾਵਰ।

ਵਰਕਰ ਨੋਡਾਂ 'ਤੇ ਸੰਖੇਪ ਜਾਣਕਾਰੀ ਵਾਲਾ ਪਹਿਲਾ ਪੰਨਾ:

ਕੰਮ ਕਰਨ ਵਾਲੇ ਕੰਮਾਂ ਦੇ ਨਾਲ ਸਭ ਤੋਂ ਤੀਬਰ ਪੰਨਾ:

ਸਾਡੇ ਬ੍ਰੋਕਰ ਦੀ ਸਥਿਤੀ ਵਾਲਾ ਸਭ ਤੋਂ ਬੋਰਿੰਗ ਪੰਨਾ:

ਸਭ ਤੋਂ ਚਮਕਦਾਰ ਪੇਜ ਟਾਸਕ ਸਟੇਟਸ ਗ੍ਰਾਫ ਅਤੇ ਉਹਨਾਂ ਦੇ ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਟਾਈਮ ਦੇ ਨਾਲ ਹੈ:

ਅਸੀਂ ਅੰਡਰਲੋਡ ਲੋਡ ਕਰਦੇ ਹਾਂ

ਇਸ ਲਈ, ਸਾਰੇ ਕੰਮ ਕੀਤੇ ਗਏ ਹਨ, ਤੁਸੀਂ ਜ਼ਖਮੀਆਂ ਨੂੰ ਚੁੱਕ ਸਕਦੇ ਹੋ.

ਅਤੇ ਬਹੁਤ ਸਾਰੇ ਜ਼ਖਮੀ ਸਨ - ਇੱਕ ਜਾਂ ਕਿਸੇ ਹੋਰ ਕਾਰਨ ਕਰਕੇ. ਏਅਰਫਲੋ ਦੀ ਸਹੀ ਵਰਤੋਂ ਦੇ ਮਾਮਲੇ ਵਿੱਚ, ਇਹ ਬਹੁਤ ਹੀ ਵਰਗ ਦਰਸਾਉਂਦੇ ਹਨ ਕਿ ਡੇਟਾ ਯਕੀਨੀ ਤੌਰ 'ਤੇ ਨਹੀਂ ਆਇਆ ਸੀ।

ਤੁਹਾਨੂੰ ਲੌਗ ਦੇਖਣ ਅਤੇ ਡਿੱਗੇ ਹੋਏ ਟਾਸਕ ਉਦਾਹਰਨਾਂ ਨੂੰ ਮੁੜ ਚਾਲੂ ਕਰਨ ਦੀ ਲੋੜ ਹੈ।

ਕਿਸੇ ਵੀ ਵਰਗ 'ਤੇ ਕਲਿੱਕ ਕਰਨ ਨਾਲ, ਅਸੀਂ ਸਾਡੇ ਲਈ ਉਪਲਬਧ ਕਾਰਵਾਈਆਂ ਦੇਖਾਂਗੇ:

ਤੁਸੀਂ ਲੈ ਸਕਦੇ ਹੋ ਅਤੇ ਕਲੀਅਰ ਦ ਫਾਲਨ ਬਣਾ ਸਕਦੇ ਹੋ। ਭਾਵ, ਅਸੀਂ ਭੁੱਲ ਜਾਂਦੇ ਹਾਂ ਕਿ ਉੱਥੇ ਕੁਝ ਅਸਫਲ ਹੋ ਗਿਆ ਹੈ, ਅਤੇ ਉਹੀ ਉਦਾਹਰਣ ਕਾਰਜ ਸ਼ਡਿਊਲਰ ਕੋਲ ਜਾਵੇਗਾ.

ਇਹ ਸਪੱਸ਼ਟ ਹੈ ਕਿ ਸਾਰੇ ਲਾਲ ਵਰਗਾਂ ਦੇ ਨਾਲ ਮਾਊਸ ਨਾਲ ਅਜਿਹਾ ਕਰਨਾ ਬਹੁਤ ਮਨੁੱਖੀ ਨਹੀਂ ਹੈ - ਇਹ ਉਹ ਨਹੀਂ ਹੈ ਜੋ ਅਸੀਂ ਏਅਰਫਲੋ ਤੋਂ ਉਮੀਦ ਕਰਦੇ ਹਾਂ. ਕੁਦਰਤੀ ਤੌਰ 'ਤੇ, ਸਾਡੇ ਕੋਲ ਵਿਆਪਕ ਤਬਾਹੀ ਦੇ ਹਥਿਆਰ ਹਨ: Browse/Task Instances

ਚਲੋ ਸਭ ਕੁਝ ਇੱਕੋ ਵਾਰ ਚੁਣੀਏ ਅਤੇ ਜ਼ੀਰੋ 'ਤੇ ਰੀਸੈਟ ਕਰੀਏ, ਸਹੀ ਆਈਟਮ 'ਤੇ ਕਲਿੱਕ ਕਰੋ:

ਸਫਾਈ ਕਰਨ ਤੋਂ ਬਾਅਦ, ਸਾਡੀਆਂ ਟੈਕਸੀਆਂ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦਿੰਦੀਆਂ ਹਨ (ਉਹ ਪਹਿਲਾਂ ਹੀ ਅਨੁਸੂਚਿਤਕਰਤਾ ਦੁਆਰਾ ਉਹਨਾਂ ਨੂੰ ਤਹਿ ਕਰਨ ਲਈ ਉਡੀਕ ਕਰ ਰਹੀਆਂ ਹਨ):

ਕਨੈਕਸ਼ਨ, ਹੁੱਕ ਅਤੇ ਹੋਰ ਵੇਰੀਏਬਲ

ਇਹ ਅਗਲੀ ਡੀਏਜੀ ਨੂੰ ਦੇਖਣ ਦਾ ਸਮਾਂ ਹੈ, update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='DWH Reports updated',
    html_content=dedent("""Господа хорошие, отчеты обновлены"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Наташ, просыпайся, мы {{ dag.dag_id }} уронили
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

ਕੀ ਹਰ ਕਿਸੇ ਨੇ ਕਦੇ ਰਿਪੋਰਟ ਅਪਡੇਟ ਕੀਤੀ ਹੈ? ਇਹ ਉਸ ਦੀ ਦੁਬਾਰਾ ਹੈ: ਇੱਥੇ ਸਰੋਤਾਂ ਦੀ ਇੱਕ ਸੂਚੀ ਹੈ ਜਿੱਥੇ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨਾ ਹੈ; ਇੱਥੇ ਇੱਕ ਸੂਚੀ ਹੈ ਜਿੱਥੇ ਲਗਾਉਣਾ ਹੈ; ਜਦੋਂ ਸਭ ਕੁਝ ਵਾਪਰਿਆ ਜਾਂ ਟੁੱਟ ਗਿਆ ਤਾਂ ਹਾਨਕ ਕਰਨਾ ਨਾ ਭੁੱਲੋ (ਠੀਕ ਹੈ, ਇਹ ਸਾਡੇ ਬਾਰੇ ਨਹੀਂ ਹੈ, ਨਹੀਂ)।

ਚਲੋ ਫਾਈਲ ਨੂੰ ਦੁਬਾਰਾ ਵੇਖੀਏ ਅਤੇ ਨਵੀਂ ਅਸਪਸ਼ਟ ਚੀਜ਼ਾਂ ਨੂੰ ਵੇਖੀਏ:

from commons.operators import TelegramBotSendMessage - ਕੋਈ ਵੀ ਚੀਜ਼ ਸਾਨੂੰ ਆਪਣੇ ਆਪਰੇਟਰ ਬਣਾਉਣ ਤੋਂ ਨਹੀਂ ਰੋਕਦੀ, ਜਿਸਦਾ ਅਸੀਂ ਅਨਬਲੌਕਡ ਨੂੰ ਸੰਦੇਸ਼ ਭੇਜਣ ਲਈ ਇੱਕ ਛੋਟਾ ਰੈਪਰ ਬਣਾ ਕੇ ਫਾਇਦਾ ਲਿਆ। (ਅਸੀਂ ਹੇਠਾਂ ਇਸ ਆਪਰੇਟਰ ਬਾਰੇ ਹੋਰ ਗੱਲ ਕਰਾਂਗੇ);
default_args={} - dag ਆਪਣੇ ਸਾਰੇ ਓਪਰੇਟਰਾਂ ਨੂੰ ਇੱਕੋ ਜਿਹੀਆਂ ਦਲੀਲਾਂ ਵੰਡ ਸਕਦਾ ਹੈ;
to='{{ var.value.all_the_kings_men }}' - ਖੇਤਰ to ਸਾਡੇ ਕੋਲ ਹਾਰਡਕੋਡ ਨਹੀਂ ਹੋਵੇਗਾ, ਪਰ ਜਿੰਜਾ ਅਤੇ ਈਮੇਲਾਂ ਦੀ ਸੂਚੀ ਦੇ ਨਾਲ ਇੱਕ ਵੇਰੀਏਬਲ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਗਤੀਸ਼ੀਲ ਤੌਰ 'ਤੇ ਤਿਆਰ ਕੀਤਾ ਗਿਆ ਹੈ, ਜੋ ਮੈਂ ਧਿਆਨ ਨਾਲ ਰੱਖਦਾ ਹਾਂ Admin/Variables;
trigger_rule=TriggerRule.ALL_SUCCESS — ਆਪਰੇਟਰ ਸ਼ੁਰੂ ਕਰਨ ਲਈ ਸ਼ਰਤ। ਸਾਡੇ ਕੇਸ ਵਿੱਚ, ਪੱਤਰ ਸਿਰਫ ਤਾਂ ਹੀ ਮਾਲਕਾਂ ਨੂੰ ਜਾਵੇਗਾ ਜੇਕਰ ਸਾਰੀਆਂ ਨਿਰਭਰਤਾਵਾਂ ਨੇ ਕੰਮ ਕੀਤਾ ਹੈ ਸਫਲਤਾਪੂਰਵਕ;
tg_bot_conn_id='tg_main' - ਦਲੀਲਾਂ conn_id ਕਨੈਕਸ਼ਨ ID ਸਵੀਕਾਰ ਕਰਦੇ ਹਾਂ ਜੋ ਅਸੀਂ ਬਣਾਉਂਦੇ ਹਾਂ Admin/Connections;
trigger_rule=TriggerRule.ONE_FAILED - ਟੈਲੀਗ੍ਰਾਮ ਵਿੱਚ ਸੁਨੇਹੇ ਤਾਂ ਹੀ ਉੱਡ ਜਾਣਗੇ ਜੇ ਕੋਈ ਕੰਮ ਡਿੱਗੇ ਹੋਏ ਹਨ;
task_concurrency=1 - ਅਸੀਂ ਇੱਕ ਕੰਮ ਦੇ ਕਈ ਟਾਸਕ ਉਦਾਹਰਨਾਂ ਨੂੰ ਇੱਕੋ ਸਮੇਂ ਲਾਂਚ ਕਰਨ ਦੀ ਮਨਾਹੀ ਕਰਦੇ ਹਾਂ। ਨਹੀਂ ਤਾਂ, ਅਸੀਂ ਕਈਆਂ ਦੀ ਇੱਕੋ ਸਮੇਂ ਲਾਂਚਿੰਗ ਪ੍ਰਾਪਤ ਕਰਾਂਗੇ VerticaOperator (ਇੱਕ ਮੇਜ਼ ਵੱਲ ਦੇਖ ਰਿਹਾ ਹੈ);
report_update >> [email, tg] - ਸਾਰੇ VerticaOperator ਚਿੱਠੀਆਂ ਅਤੇ ਸੁਨੇਹਿਆਂ ਨੂੰ ਭੇਜਣ ਵਿੱਚ ਇਕਸਾਰ ਹੋਵੋ, ਇਸ ਤਰ੍ਹਾਂ:

ਪਰ ਕਿਉਂਕਿ ਨੋਟੀਫਾਇਰ ਆਪਰੇਟਰਾਂ ਦੀਆਂ ਵੱਖ-ਵੱਖ ਲਾਂਚ ਸ਼ਰਤਾਂ ਹਨ, ਕੇਵਲ ਇੱਕ ਹੀ ਕੰਮ ਕਰੇਗਾ। ਟ੍ਰੀ ਵਿਊ ਵਿੱਚ, ਸਭ ਕੁਝ ਥੋੜਾ ਘੱਟ ਵਿਜ਼ੂਅਲ ਦਿਖਾਈ ਦਿੰਦਾ ਹੈ:

ਬਾਰੇ ਕੁਝ ਸ਼ਬਦ ਕਹਾਂਗਾ ਮੈਕਰੋ ਅਤੇ ਉਹਨਾਂ ਦੇ ਦੋਸਤ - ਵੇਰੀਏਬਲ.

ਮੈਕਰੋ ਜਿੰਜਾ ਪਲੇਸਹੋਲਡਰ ਹਨ ਜੋ ਵੱਖ-ਵੱਖ ਉਪਯੋਗੀ ਜਾਣਕਾਰੀ ਨੂੰ ਆਪਰੇਟਰ ਆਰਗੂਮੈਂਟਾਂ ਵਿੱਚ ਬਦਲ ਸਕਦੇ ਹਨ। ਉਦਾਹਰਨ ਲਈ, ਇਸ ਤਰ੍ਹਾਂ:

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} ਸੰਦਰਭ ਵੇਰੀਏਬਲ ਦੀ ਸਮੱਗਰੀ ਤੱਕ ਵਿਸਤਾਰ ਕਰੇਗਾ execution_date ਫਾਰਮੈਟ ਵਿੱਚ YYYY-MM-DD: 2020-07-14. ਸਭ ਤੋਂ ਵਧੀਆ ਗੱਲ ਇਹ ਹੈ ਕਿ ਸੰਦਰਭ ਵੇਰੀਏਬਲਾਂ ਨੂੰ ਇੱਕ ਖਾਸ ਟਾਸਕ ਉਦਾਹਰਨ (ਟ੍ਰੀ ਵਿਊ ਵਿੱਚ ਇੱਕ ਵਰਗ) ਨਾਲ ਜੋੜਿਆ ਜਾਂਦਾ ਹੈ, ਅਤੇ ਜਦੋਂ ਮੁੜ ਚਾਲੂ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਤਾਂ ਪਲੇਸਹੋਲਡਰ ਉਸੇ ਮੁੱਲਾਂ ਤੱਕ ਫੈਲ ਜਾਣਗੇ।

ਨਿਰਧਾਰਤ ਮੁੱਲਾਂ ਨੂੰ ਹਰੇਕ ਟਾਸਕ ਉਦਾਹਰਨ 'ਤੇ ਰੈਂਡਰਡ ਬਟਨ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਦੇਖਿਆ ਜਾ ਸਕਦਾ ਹੈ। ਪੱਤਰ ਭੇਜਣ ਦਾ ਕੰਮ ਇਸ ਤਰ੍ਹਾਂ ਹੈ:

ਅਤੇ ਇਸ ਲਈ ਇੱਕ ਸੁਨੇਹਾ ਭੇਜਣ ਦੇ ਕੰਮ 'ਤੇ:

ਨਵੀਨਤਮ ਉਪਲਬਧ ਸੰਸਕਰਣ ਲਈ ਬਿਲਟ-ਇਨ ਮੈਕਰੋ ਦੀ ਇੱਕ ਪੂਰੀ ਸੂਚੀ ਇੱਥੇ ਉਪਲਬਧ ਹੈ: ਮੈਕਰੋ ਸੰਦਰਭ

ਇਸ ਤੋਂ ਇਲਾਵਾ, ਪਲੱਗਇਨ ਦੀ ਮਦਦ ਨਾਲ, ਅਸੀਂ ਆਪਣੇ ਖੁਦ ਦੇ ਮੈਕਰੋ ਘੋਸ਼ਿਤ ਕਰ ਸਕਦੇ ਹਾਂ, ਪਰ ਇਹ ਇਕ ਹੋਰ ਕਹਾਣੀ ਹੈ।

ਪਹਿਲਾਂ ਤੋਂ ਪਰਿਭਾਸ਼ਿਤ ਚੀਜ਼ਾਂ ਤੋਂ ਇਲਾਵਾ, ਅਸੀਂ ਆਪਣੇ ਵੇਰੀਏਬਲ ਦੇ ਮੁੱਲਾਂ ਨੂੰ ਬਦਲ ਸਕਦੇ ਹਾਂ (ਮੈਂ ਪਹਿਲਾਂ ਹੀ ਉਪਰੋਕਤ ਕੋਡ ਵਿੱਚ ਇਸਦੀ ਵਰਤੋਂ ਕੀਤੀ ਹੈ)। ਚਲੋ ਅੰਦਰ ਬਣਾਉ Admin/Variables ਕੁਝ ਚੀਜ਼ਾਂ:

ਹਰ ਚੀਜ਼ ਜੋ ਤੁਸੀਂ ਵਰਤ ਸਕਦੇ ਹੋ:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

ਮੁੱਲ ਇੱਕ ਸਕੇਲਰ ਹੋ ਸਕਦਾ ਹੈ, ਜਾਂ ਇਹ JSON ਵੀ ਹੋ ਸਕਦਾ ਹੈ। JSON ਦੇ ਮਾਮਲੇ ਵਿੱਚ:

bot_config

{
    "bot": {
        "token": 881hskdfASDA16641,
        "name": "Verter"
    },
    "service": "TG"
}

ਸਿਰਫ਼ ਲੋੜੀਂਦੀ ਕੁੰਜੀ ਲਈ ਮਾਰਗ ਦੀ ਵਰਤੋਂ ਕਰੋ: {{ var.json.bot_config.bot.token }}.

ਮੈਂ ਸ਼ਾਬਦਿਕ ਤੌਰ 'ਤੇ ਇੱਕ ਸ਼ਬਦ ਕਹਾਂਗਾ ਅਤੇ ਇਸ ਬਾਰੇ ਇੱਕ ਸਕ੍ਰੀਨਸ਼ੌਟ ਦਿਖਾਵਾਂਗਾ ਕੁਨੈਕਸ਼ਨ. ਇੱਥੇ ਸਭ ਕੁਝ ਮੁੱਢਲੀ ਹੈ: ਪੰਨੇ 'ਤੇ Admin/Connections ਅਸੀਂ ਇੱਕ ਕੁਨੈਕਸ਼ਨ ਬਣਾਉਂਦੇ ਹਾਂ, ਉੱਥੇ ਸਾਡੇ ਲੌਗਿਨ/ਪਾਸਵਰਡ ਅਤੇ ਹੋਰ ਖਾਸ ਮਾਪਦੰਡ ਜੋੜਦੇ ਹਾਂ। ਇਸ ਤਰ੍ਹਾਂ:

ਪਾਸਵਰਡਾਂ ਨੂੰ ਐਨਕ੍ਰਿਪਟ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ (ਡਿਫੌਲਟ ਨਾਲੋਂ ਜ਼ਿਆਦਾ ਚੰਗੀ ਤਰ੍ਹਾਂ), ਜਾਂ ਤੁਸੀਂ ਕੁਨੈਕਸ਼ਨ ਦੀ ਕਿਸਮ ਨੂੰ ਛੱਡ ਸਕਦੇ ਹੋ (ਜਿਵੇਂ ਕਿ ਮੈਂ ਇਸ ਲਈ ਕੀਤਾ ਸੀ। tg_main) - ਤੱਥ ਇਹ ਹੈ ਕਿ ਕਿਸਮਾਂ ਦੀ ਸੂਚੀ ਏਅਰਫਲੋ ਮਾਡਲਾਂ ਵਿੱਚ ਹਾਰਡਵਾਇਰਡ ਹੈ ਅਤੇ ਸਰੋਤ ਕੋਡਾਂ ਵਿੱਚ ਸ਼ਾਮਲ ਕੀਤੇ ਬਿਨਾਂ ਵਿਸਤਾਰ ਨਹੀਂ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ (ਜੇ ਅਚਾਨਕ ਮੈਂ ਕੁਝ ਗੂਗਲ ਨਹੀਂ ਕੀਤਾ, ਤਾਂ ਕਿਰਪਾ ਕਰਕੇ ਮੈਨੂੰ ਠੀਕ ਕਰੋ), ਪਰ ਕੁਝ ਵੀ ਸਾਨੂੰ ਕ੍ਰੈਡਿਟ ਪ੍ਰਾਪਤ ਕਰਨ ਤੋਂ ਰੋਕ ਨਹੀਂ ਸਕੇਗਾ ਨਾਮ

ਤੁਸੀਂ ਇੱਕੋ ਨਾਮ ਦੇ ਨਾਲ ਕਈ ਕੁਨੈਕਸ਼ਨ ਵੀ ਬਣਾ ਸਕਦੇ ਹੋ: ਇਸ ਕੇਸ ਵਿੱਚ, ਵਿਧੀ BaseHook.get_connection(), ਜੋ ਸਾਨੂੰ ਨਾਮ ਦੁਆਰਾ ਕੁਨੈਕਸ਼ਨ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹੈ, ਦੇਵੇਗਾ ਬੇਤਰਤੀਬ ਕਈ ਨਾਮਾਂ ਤੋਂ (ਰਾਊਂਡ ਰੌਬਿਨ ਬਣਾਉਣਾ ਵਧੇਰੇ ਤਰਕਪੂਰਨ ਹੋਵੇਗਾ, ਪਰ ਆਓ ਇਸਨੂੰ ਏਅਰਫਲੋ ਡਿਵੈਲਪਰਾਂ ਦੀ ਜ਼ਮੀਰ 'ਤੇ ਛੱਡ ਦੇਈਏ)।

ਵੇਰੀਏਬਲ ਅਤੇ ਕਨੈਕਸ਼ਨ ਨਿਸ਼ਚਿਤ ਤੌਰ 'ਤੇ ਵਧੀਆ ਟੂਲ ਹਨ, ਪਰ ਸੰਤੁਲਨ ਨਾ ਗੁਆਉਣਾ ਮਹੱਤਵਪੂਰਨ ਹੈ: ਤੁਹਾਡੇ ਪ੍ਰਵਾਹ ਦੇ ਕਿਹੜੇ ਹਿੱਸੇ ਤੁਸੀਂ ਕੋਡ ਵਿੱਚ ਸਟੋਰ ਕਰਦੇ ਹੋ, ਅਤੇ ਤੁਸੀਂ ਸਟੋਰੇਜ ਲਈ ਏਅਰਫਲੋ ਨੂੰ ਕਿਹੜੇ ਹਿੱਸੇ ਦਿੰਦੇ ਹੋ। ਇੱਕ ਪਾਸੇ, UI ਦੁਆਰਾ ਮੁੱਲ ਨੂੰ ਤੇਜ਼ੀ ਨਾਲ ਬਦਲਣਾ ਸੁਵਿਧਾਜਨਕ ਹੋ ਸਕਦਾ ਹੈ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ ਮੇਲਿੰਗ ਬਾਕਸ। ਦੂਜੇ ਪਾਸੇ, ਇਹ ਅਜੇ ਵੀ ਮਾਊਸ ਕਲਿੱਕ ਦੀ ਵਾਪਸੀ ਹੈ, ਜਿਸ ਤੋਂ ਅਸੀਂ (ਮੈਂ) ਛੁਟਕਾਰਾ ਪਾਉਣਾ ਚਾਹੁੰਦੇ ਸੀ।

ਕੁਨੈਕਸ਼ਨਾਂ ਨਾਲ ਕੰਮ ਕਰਨਾ ਇੱਕ ਕੰਮ ਹੈ ਹੁੱਕ. ਆਮ ਤੌਰ 'ਤੇ, ਏਅਰਫਲੋ ਹੁੱਕ ਇਸ ਨੂੰ ਤੀਜੀ-ਧਿਰ ਦੀਆਂ ਸੇਵਾਵਾਂ ਅਤੇ ਲਾਇਬ੍ਰੇਰੀਆਂ ਨਾਲ ਜੋੜਨ ਲਈ ਪੁਆਇੰਟ ਹੁੰਦੇ ਹਨ। ਜਿਵੇਂ, JiraHook ਜੀਰਾ ਨਾਲ ਗੱਲਬਾਤ ਕਰਨ ਲਈ ਸਾਡੇ ਲਈ ਇੱਕ ਕਲਾਇੰਟ ਖੋਲ੍ਹੇਗਾ (ਤੁਸੀਂ ਕੰਮਾਂ ਨੂੰ ਅੱਗੇ ਅਤੇ ਪਿੱਛੇ ਕਰ ਸਕਦੇ ਹੋ), ਅਤੇ ਦੀ ਮਦਦ ਨਾਲ SambaHook ਤੁਸੀਂ ਇੱਕ ਸਥਾਨਕ ਫਾਈਲ ਨੂੰ ਦਬਾ ਸਕਦੇ ਹੋ smb- ਬਿੰਦੂ.

ਕਸਟਮ ਆਪਰੇਟਰ ਨੂੰ ਪਾਰਸ ਕਰ ਰਿਹਾ ਹੈ

ਅਤੇ ਅਸੀਂ ਇਹ ਦੇਖਣ ਦੇ ਨੇੜੇ ਹੋ ਗਏ ਕਿ ਇਹ ਕਿਵੇਂ ਬਣਿਆ ਹੈ TelegramBotSendMessage

ਕੋਡ commons/operators.py ਅਸਲ ਆਪਰੇਟਰ ਨਾਲ:

from typing import Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Send message to chat_id using TelegramBotHook

    Example:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Union[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Send "{self.message}" to the chat {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

ਇੱਥੇ, ਏਅਰਫਲੋ ਵਿੱਚ ਹਰ ਚੀਜ਼ ਦੀ ਤਰ੍ਹਾਂ, ਹਰ ਚੀਜ਼ ਬਹੁਤ ਸਧਾਰਨ ਹੈ:

ਤੋਂ ਵਿਰਸੇ ਵਿੱਚ ਮਿਲੀ BaseOperator, ਜੋ ਕਿ ਕੁਝ ਏਅਰਫਲੋ-ਵਿਸ਼ੇਸ਼ ਚੀਜ਼ਾਂ ਨੂੰ ਲਾਗੂ ਕਰਦਾ ਹੈ (ਆਪਣੇ ਮਨੋਰੰਜਨ ਨੂੰ ਦੇਖੋ)
ਘੋਸ਼ਿਤ ਖੇਤਰ template_fields, ਜਿਸ ਵਿੱਚ Jinja ਪ੍ਰਕਿਰਿਆ ਕਰਨ ਲਈ ਮੈਕਰੋ ਦੀ ਖੋਜ ਕਰੇਗਾ।
ਲਈ ਸਹੀ ਦਲੀਲਾਂ ਦਾ ਪ੍ਰਬੰਧ ਕੀਤਾ __init__(), ਜਿੱਥੇ ਲੋੜ ਹੋਵੇ ਡਿਫਾਲਟ ਸੈੱਟ ਕਰੋ।
ਅਸੀਂ ਪੂਰਵਜ ਦੀ ਸ਼ੁਰੂਆਤ ਬਾਰੇ ਵੀ ਨਹੀਂ ਭੁੱਲੇ।
ਅਨੁਸਾਰੀ ਹੁੱਕ ਖੋਲ੍ਹੀ TelegramBotHookਇਸ ਤੋਂ ਇੱਕ ਕਲਾਇੰਟ ਆਬਜੈਕਟ ਪ੍ਰਾਪਤ ਕੀਤਾ।
ਓਵਰਰਾਈਡ (ਮੁੜ ਪਰਿਭਾਸ਼ਿਤ) ਵਿਧੀ BaseOperator.execute(), ਜਦੋਂ ਓਪਰੇਟਰ ਨੂੰ ਲਾਂਚ ਕਰਨ ਦਾ ਸਮਾਂ ਆਉਂਦਾ ਹੈ ਤਾਂ ਏਅਰਫੋ ਨੂੰ ਮਰੋੜ ਦੇਵੇਗਾ - ਇਸ ਵਿੱਚ ਅਸੀਂ ਲੌਗਇਨ ਕਰਨਾ ਭੁੱਲ ਕੇ ਮੁੱਖ ਕਾਰਵਾਈ ਨੂੰ ਲਾਗੂ ਕਰਾਂਗੇ। (ਅਸੀਂ ਲੌਗਇਨ ਕਰਦੇ ਹਾਂ, ਤਰੀਕੇ ਨਾਲ, ਬਿਲਕੁਲ ਅੰਦਰ stdout и stderr - ਏਅਰਫਲੋ ਹਰ ਚੀਜ਼ ਨੂੰ ਰੋਕ ਦੇਵੇਗਾ, ਇਸ ਨੂੰ ਸੁੰਦਰਤਾ ਨਾਲ ਲਪੇਟ ਦੇਵੇਗਾ, ਜਿੱਥੇ ਲੋੜ ਹੋਵੇ, ਇਸ ਨੂੰ ਵਿਗਾੜ ਦੇਵੇਗਾ।)

ਆਓ ਦੇਖੀਏ ਕਿ ਸਾਡੇ ਕੋਲ ਕੀ ਹੈ commons/hooks.py. ਫਾਈਲ ਦਾ ਪਹਿਲਾ ਹਿੱਸਾ, ਹੁੱਕ ਦੇ ਨਾਲ:

from typing import Union

from airflow.hooks.base_hook import BaseHook
from requests_toolbelt.sessions import BaseUrlSession

class TelegramBotHook(BaseHook):
    """Telegram Bot API hook

    Note: add a connection with empty Conn Type and don't forget
    to fill Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

ਮੈਨੂੰ ਇਹ ਵੀ ਨਹੀਂ ਪਤਾ ਕਿ ਇੱਥੇ ਕੀ ਸਮਝਾਉਣਾ ਹੈ, ਮੈਂ ਸਿਰਫ਼ ਮਹੱਤਵਪੂਰਨ ਨੁਕਤੇ ਨੋਟ ਕਰਾਂਗਾ:

ਅਸੀਂ ਵਿਰਾਸਤ ਵਿੱਚ ਹਾਂ, ਦਲੀਲਾਂ ਬਾਰੇ ਸੋਚੋ - ਜ਼ਿਆਦਾਤਰ ਮਾਮਲਿਆਂ ਵਿੱਚ ਇਹ ਇੱਕ ਹੋਵੇਗਾ: conn_id;
ਮਿਆਰੀ ਢੰਗਾਂ ਨੂੰ ਓਵਰਰਾਈਡ ਕਰਨਾ: ਮੈਂ ਆਪਣੇ ਆਪ ਨੂੰ ਸੀਮਤ ਕਰ ਲਿਆ ਹੈ get_conn(), ਜਿਸ ਵਿੱਚ ਮੈਂ ਨਾਮ ਦੁਆਰਾ ਕਨੈਕਸ਼ਨ ਪੈਰਾਮੀਟਰ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹਾਂ ਅਤੇ ਕੇਵਲ ਭਾਗ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹਾਂ extra (ਇਹ ਇੱਕ JSON ਖੇਤਰ ਹੈ), ਜਿਸ ਵਿੱਚ ਮੈਂ (ਮੇਰੀਆਂ ਆਪਣੀਆਂ ਹਦਾਇਤਾਂ ਅਨੁਸਾਰ!) ਟੈਲੀਗ੍ਰਾਮ ਬੋਟ ਟੋਕਨ ਪਾ ਦਿੱਤਾ ਹੈ: {"bot_token": "YOuRAwEsomeBOtToKen"}.
ਮੈਂ ਸਾਡੀ ਇੱਕ ਉਦਾਹਰਣ ਬਣਾਉਂਦਾ ਹਾਂ TelegramBot, ਇਸ ਨੂੰ ਇੱਕ ਖਾਸ ਟੋਕਨ ਦਿੰਦੇ ਹੋਏ।

ਇਹ ਸਭ ਹੈ. ਤੁਸੀਂ ਵਰਤ ਕੇ ਇੱਕ ਹੁੱਕ ਤੋਂ ਇੱਕ ਕਲਾਇੰਟ ਪ੍ਰਾਪਤ ਕਰ ਸਕਦੇ ਹੋ TelegramBotHook().clent ਜ TelegramBotHook().get_conn().

ਅਤੇ ਫਾਈਲ ਦਾ ਦੂਜਾ ਹਿੱਸਾ, ਜਿਸ ਵਿੱਚ ਮੈਂ ਟੈਲੀਗ੍ਰਾਮ REST API ਲਈ ਇੱਕ ਮਾਈਕ੍ਰੋਵਰੈਪਰ ਬਣਾਉਂਦਾ ਹਾਂ, ਤਾਂ ਜੋ ਉਸੇ ਨੂੰ ਖਿੱਚਿਆ ਨਾ ਜਾਵੇ python-telegram-bot ਇੱਕ ਢੰਗ ਲਈ sendMessage.

class TelegramBot:
    """Telegram Bot API wrapper

    Examples:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hi, darling')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hi, darling', chat_id=-1762374628374)
    """
    API_ENDPOINT = 'https://api.telegram.org/bot{}/'

    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):
        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)
        self.session = BaseUrlSession(self._base_url)
        self.chat_id = chat_id

    def send_message(self, message: str, chat_id: Union[int, str] = None):
        method = 'sendMessage'

        payload = {'chat_id': chat_id or self.chat_id,
                   'text': message,
                   'parse_mode': 'MarkdownV2'}

        response = self.session.post(method, data=payload).json()
        if not response.get('ok'):
            raise TelegramBotException(response)

class TelegramBotException(Exception):
    def __init__(self, *args, **kwargs):
        super().__init__((args, kwargs))

ਸਹੀ ਤਰੀਕਾ ਹੈ ਇਹ ਸਭ ਜੋੜਨਾ: TelegramBotSendMessage, TelegramBotHook, TelegramBot - ਪਲੱਗਇਨ ਵਿੱਚ, ਇੱਕ ਜਨਤਕ ਰਿਪੋਜ਼ਟਰੀ ਵਿੱਚ ਪਾਓ, ਅਤੇ ਇਸਨੂੰ ਓਪਨ ਸੋਰਸ ਨੂੰ ਦਿਓ।

ਜਦੋਂ ਅਸੀਂ ਇਸ ਸਭ ਦਾ ਅਧਿਐਨ ਕਰ ਰਹੇ ਸੀ, ਸਾਡੀ ਰਿਪੋਰਟ ਅੱਪਡੇਟ ਸਫਲਤਾਪੂਰਵਕ ਅਸਫਲ ਹੋ ਗਈ ਅਤੇ ਮੈਨੂੰ ਚੈਨਲ ਵਿੱਚ ਇੱਕ ਗਲਤੀ ਸੁਨੇਹਾ ਭੇਜਿਆ ਗਿਆ। ਮੈਂ ਇਹ ਦੇਖਣ ਲਈ ਜਾਂਚ ਕਰਨ ਜਾ ਰਿਹਾ ਹਾਂ ਕਿ ਕੀ ਇਹ ਗਲਤ ਹੈ...

ਸਾਡੇ ਕੁੱਤੇ ਵਿੱਚ ਕੁਝ ਟੁੱਟ ਗਿਆ! ਕੀ ਇਹ ਉਹੀ ਨਹੀਂ ਹੈ ਜਿਸਦੀ ਅਸੀਂ ਉਮੀਦ ਕਰ ਰਹੇ ਸੀ? ਬਿਲਕੁਲ!

ਕੀ ਤੁਸੀਂ ਡੋਲ੍ਹਣ ਜਾ ਰਹੇ ਹੋ?

ਕੀ ਤੁਸੀਂ ਮਹਿਸੂਸ ਕਰਦੇ ਹੋ ਕਿ ਮੈਂ ਕੁਝ ਗੁਆ ਦਿੱਤਾ ਹੈ? ਅਜਿਹਾ ਲਗਦਾ ਹੈ ਕਿ ਉਸਨੇ SQL ਸਰਵਰ ਤੋਂ ਵਰਟਿਕਾ ਵਿੱਚ ਡੇਟਾ ਟ੍ਰਾਂਸਫਰ ਕਰਨ ਦਾ ਵਾਅਦਾ ਕੀਤਾ ਸੀ, ਅਤੇ ਫਿਰ ਉਸਨੇ ਇਸਨੂੰ ਲਿਆ ਅਤੇ ਵਿਸ਼ੇ ਨੂੰ ਛੱਡ ਦਿੱਤਾ, ਬਦਨਾਮ!

ਇਹ ਅੱਤਿਆਚਾਰ ਜਾਣਬੁੱਝ ਕੇ ਕੀਤਾ ਗਿਆ ਸੀ, ਮੈਨੂੰ ਤੁਹਾਡੇ ਲਈ ਕੁਝ ਸ਼ਬਦਾਵਲੀ ਸਮਝਣੀ ਸੀ। ਹੁਣ ਤੁਸੀਂ ਹੋਰ ਅੱਗੇ ਜਾ ਸਕਦੇ ਹੋ।

ਸਾਡੀ ਯੋਜਨਾ ਇਹ ਸੀ:

ਡਾਗ ਕਰੋ
ਕਾਰਜ ਤਿਆਰ ਕਰੋ
ਦੇਖੋ ਕਿ ਹਰ ਚੀਜ਼ ਕਿੰਨੀ ਸੁੰਦਰ ਹੈ
ਭਰਨ ਲਈ ਸੈਸ਼ਨ ਨੰਬਰ ਨਿਰਧਾਰਤ ਕਰੋ
SQL ਸਰਵਰ ਤੋਂ ਡਾਟਾ ਪ੍ਰਾਪਤ ਕਰੋ
Vertica ਵਿੱਚ ਡੇਟਾ ਪਾਓ
ਅੰਕੜੇ ਇਕੱਠੇ ਕਰੋ

ਇਸ ਲਈ, ਇਸ ਸਭ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰਨ ਅਤੇ ਚਲਾਉਣ ਲਈ, ਮੈਂ ਸਾਡੇ ਵਿੱਚ ਇੱਕ ਛੋਟਾ ਜਿਹਾ ਵਾਧਾ ਕੀਤਾ docker-compose.yml:

docker-compose.db.yml

version: '3.4'

x-mssql-base: &mssql-base
  image: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04
  restart: always
  environment:
    ACCEPT_EULA: Y
    MSSQL_PID: Express
    SA_PASSWORD: SayThanksToSatiaAt2020
    MSSQL_MEMORY_LIMIT_MB: 1024

services:
  dwh:
    image: jbfavre/vertica:9.2.0-7_ubuntu-16.04

  mssql_0:
    <<: *mssql-base

  mssql_1:
    <<: *mssql-base

  mssql_2:
    <<: *mssql-base

  mssql_init:
    image: mio101/py3-sql-db-client-base
    command: python3 ./mssql_init.py
    depends_on:
      - mssql_0
      - mssql_1
      - mssql_2
    environment:
      SA_PASSWORD: SayThanksToSatiaAt2020
    volumes:
      - ./mssql_init.py:/mssql_init.py
      - ./dags/commons/datasources.py:/commons/datasources.py

ਉੱਥੇ ਅਸੀਂ ਉਠਾਉਂਦੇ ਹਾਂ:

ਮੇਜ਼ਬਾਨ ਵਜੋਂ Vertica dwh ਸਭ ਤੋਂ ਡਿਫੌਲਟ ਸੈਟਿੰਗਾਂ ਦੇ ਨਾਲ,
SQL ਸਰਵਰ ਦੇ ਤਿੰਨ ਮੌਕੇ,
ਅਸੀਂ ਬਾਅਦ ਵਿੱਚ ਡੇਟਾਬੇਸ ਨੂੰ ਕੁਝ ਡੇਟਾ ਨਾਲ ਭਰਦੇ ਹਾਂ (ਕਿਸੇ ਵੀ ਸਥਿਤੀ ਵਿੱਚ ਨਾ ਵੇਖੋ mssql_init.py!)

ਅਸੀਂ ਪਿਛਲੀ ਵਾਰ ਦੇ ਮੁਕਾਬਲੇ ਥੋੜ੍ਹੀ ਜਿਹੀ ਗੁੰਝਲਦਾਰ ਕਮਾਂਡ ਦੀ ਮਦਦ ਨਾਲ ਸਾਰੇ ਚੰਗੇ ਲਾਂਚ ਕਰਦੇ ਹਾਂ:

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

ਸਾਡੇ ਚਮਤਕਾਰ ਰੈਂਡਮਾਈਜ਼ਰ ਨੇ ਕੀ ਬਣਾਇਆ ਹੈ, ਤੁਸੀਂ ਆਈਟਮ ਦੀ ਵਰਤੋਂ ਕਰ ਸਕਦੇ ਹੋ Data Profiling/Ad Hoc Query:

ਮੁੱਖ ਗੱਲ ਇਹ ਹੈ ਕਿ ਇਸ ਨੂੰ ਵਿਸ਼ਲੇਸ਼ਕਾਂ ਨੂੰ ਦਿਖਾਉਣਾ ਨਹੀਂ ਹੈ

'ਤੇ ਵਿਸਤ੍ਰਿਤ ETL ਸੈਸ਼ਨ ਮੈਂ ਨਹੀਂ ਕਰਾਂਗਾ, ਇੱਥੇ ਸਭ ਕੁਝ ਮਾਮੂਲੀ ਹੈ: ਅਸੀਂ ਇੱਕ ਅਧਾਰ ਬਣਾਉਂਦੇ ਹਾਂ, ਇਸ ਵਿੱਚ ਇੱਕ ਚਿੰਨ੍ਹ ਹੁੰਦਾ ਹੈ, ਅਸੀਂ ਇੱਕ ਪ੍ਰਸੰਗ ਮੈਨੇਜਰ ਨਾਲ ਹਰ ਚੀਜ਼ ਨੂੰ ਸਮੇਟਦੇ ਹਾਂ, ਅਤੇ ਹੁਣ ਅਸੀਂ ਇਹ ਕਰਦੇ ਹਾਂ:

with Session(task_name) as session:
    print('Load', session.id, 'started')

    # Load worflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

from sys import stderr

class Session:
    """ETL workflow session

    Example:
        with Session(task_name) as session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Well done'
    """

    def __init__(self, connection, task_name):
        self.connection = connection
        self.connection.autocommit = True

        self._task_name = task_name
        self._id = None

        self.loaded_rows = None
        self.successful = None
        self.comment = None

    def __enter__(self):
        return self.open()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if any(exc_type, exc_val, exc_tb):
            self.successful = False
            self.comment = f'{exc_type}: {exc_val}n{exc_tb}'
            print(exc_type, exc_val, exc_tb, file=stderr)
        self.close()

    def __repr__(self):
        return (f'<{self.__class__.__name__} ' 
                f'id={self.id} ' 
                f'task_name="{self.task_name}">')

    @property
    def task_name(self):
        return self._task_name

    @property
    def id(self):
        return self._id

    def _execute(self, query, *args):
        with self.connection.cursor() as cursor:
            cursor.execute(query, args)
            return cursor.fetchone()[0]

    def _create(self):
        query = """
            CREATE TABLE IF NOT EXISTS sessions (
                id          SERIAL       NOT NULL PRIMARY KEY,
                task_name   VARCHAR(200) NOT NULL,

                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,
                finished    TIMESTAMPTZ           DEFAULT current_timestamp,
                successful  BOOL,

                loaded_rows INT,
                comment     VARCHAR(500)
            );
            """
        self._execute(query)

    def open(self):
        query = """
            INSERT INTO sessions (task_name, finished)
            VALUES (%s, NULL)
            RETURNING id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'opened')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Session is not open')
        query = """
            UPDATE sessions
            SET
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            WHERE
                id = %s
            RETURNING id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'closed',
              ', successful: ', self.successful,
              ', Loaded: ', self.loaded_rows,
              ', comment:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

ਸਮਾਂ ਆ ਗਿਆ ਹੈ ਸਾਡਾ ਡਾਟਾ ਇਕੱਠਾ ਕਰੋ ਸਾਡੇ ਡੇਢ ਸੌ ਮੇਜ਼ਾਂ ਤੋਂ। ਆਉ ਇਸਨੂੰ ਬਹੁਤ ਹੀ ਬੇਮਿਸਾਲ ਲਾਈਨਾਂ ਦੀ ਮਦਦ ਨਾਲ ਕਰੀਏ:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()

query = f"""
    SELECT 
        id, start_time, end_time, type, data
    FROM dbo.Orders
    WHERE
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)

ਹੁੱਕ ਦੀ ਮਦਦ ਨਾਲ ਅਸੀਂ ਏਅਰਫਲੋ ਤੋਂ ਪ੍ਰਾਪਤ ਕਰਦੇ ਹਾਂ pymssql- ਜੁੜੋ
ਆਉ ਬੇਨਤੀ ਵਿੱਚ ਇੱਕ ਮਿਤੀ ਦੇ ਰੂਪ ਵਿੱਚ ਇੱਕ ਪਾਬੰਦੀ ਨੂੰ ਬਦਲੀਏ - ਇਸਨੂੰ ਟੈਂਪਲੇਟ ਇੰਜਣ ਦੁਆਰਾ ਫੰਕਸ਼ਨ ਵਿੱਚ ਸੁੱਟ ਦਿੱਤਾ ਜਾਵੇਗਾ।
ਸਾਡੀ ਬੇਨਤੀ ਨੂੰ ਖੁਆਉਣਾ pandasਜੋ ਸਾਨੂੰ ਪ੍ਰਾਪਤ ਕਰੇਗਾ DataFrame - ਇਹ ਭਵਿੱਖ ਵਿੱਚ ਸਾਡੇ ਲਈ ਲਾਭਦਾਇਕ ਹੋਵੇਗਾ।

ਮੈਂ ਬਦਲ ਦੀ ਵਰਤੋਂ ਕਰ ਰਿਹਾ ਹਾਂ {dt} ਇੱਕ ਬੇਨਤੀ ਪੈਰਾਮੀਟਰ ਦੀ ਬਜਾਏ %s ਇਸ ਲਈ ਨਹੀਂ ਕਿ ਮੈਂ ਇੱਕ ਦੁਸ਼ਟ ਪਿਨੋਚਿਓ ਹਾਂ, ਪਰ ਕਿਉਂਕਿ pandas ਸੰਭਾਲ ਨਹੀਂ ਸਕਦਾ pymssql ਅਤੇ ਆਖਰੀ ਨੂੰ ਖਿਸਕਾਉਂਦਾ ਹੈ params: Listਹਾਲਾਂਕਿ ਉਹ ਅਸਲ ਵਿੱਚ ਚਾਹੁੰਦਾ ਹੈ tuple.
ਇਹ ਵੀ ਨੋਟ ਕਰੋ ਕਿ ਡਿਵੈਲਪਰ pymssql ਹੁਣ ਉਸ ਦਾ ਸਮਰਥਨ ਨਾ ਕਰਨ ਦਾ ਫੈਸਲਾ ਕੀਤਾ, ਅਤੇ ਇਹ ਬਾਹਰ ਜਾਣ ਦਾ ਸਮਾਂ ਹੈ pyodbc.

ਆਓ ਦੇਖੀਏ ਕਿ ਏਅਰਫਲੋ ਨੇ ਸਾਡੇ ਫੰਕਸ਼ਨਾਂ ਦੀਆਂ ਦਲੀਲਾਂ ਨੂੰ ਕਿਸ ਨਾਲ ਭਰਿਆ ਹੈ:

ਜੇਕਰ ਕੋਈ ਡਾਟਾ ਨਹੀਂ ਹੈ, ਤਾਂ ਜਾਰੀ ਰੱਖਣ ਦਾ ਕੋਈ ਮਤਲਬ ਨਹੀਂ ਹੈ. ਪਰ ਭਰਨ ਨੂੰ ਸਫਲ ਮੰਨਣਾ ਵੀ ਅਜੀਬ ਹੈ। ਪਰ ਇਹ ਕੋਈ ਗਲਤੀ ਨਹੀਂ ਹੈ। A-ah-ah, ਕੀ ਕਰਨਾ ਹੈ ?! ਅਤੇ ਇੱਥੇ ਕੀ ਹੈ:

if df.empty:
    raise AirflowSkipException('No rows to load')

AirflowSkipException ਏਅਰਫਲੋ ਨੂੰ ਦੱਸਦਾ ਹੈ ਕਿ ਕੋਈ ਤਰੁੱਟੀਆਂ ਨਹੀਂ ਹਨ, ਪਰ ਅਸੀਂ ਕੰਮ ਨੂੰ ਛੱਡ ਦਿੰਦੇ ਹਾਂ। ਇੰਟਰਫੇਸ ਵਿੱਚ ਇੱਕ ਹਰਾ ਜਾਂ ਲਾਲ ਵਰਗ ਨਹੀਂ ਹੋਵੇਗਾ, ਪਰ ਗੁਲਾਬੀ.

ਆਉ ਸਾਡੇ ਡੇਟਾ ਨੂੰ ਟੌਸ ਕਰੀਏ ਕਈ ਕਾਲਮ:

df['etl_source'] = src_schema
df['etl_id'] = session.id
df['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

ਅਰਥਾਤ:

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

ਅੰਤਮ ਪੜਾਅ ਬਾਕੀ ਹੈ: ਹਰ ਚੀਜ਼ ਨੂੰ ਵਰਟੀਕਾ ਵਿੱਚ ਡੋਲ੍ਹ ਦਿਓ। ਅਤੇ, ਅਜੀਬ ਤੌਰ 'ਤੇ, ਅਜਿਹਾ ਕਰਨ ਦੇ ਸਭ ਤੋਂ ਸ਼ਾਨਦਾਰ ਅਤੇ ਕੁਸ਼ਲ ਤਰੀਕਿਆਂ ਵਿੱਚੋਂ ਇੱਕ ਹੈ CSV ਦੁਆਰਾ!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

ਅਸੀਂ ਇੱਕ ਵਿਸ਼ੇਸ਼ ਰਿਸੀਵਰ ਬਣਾ ਰਹੇ ਹਾਂ StringIO.
pandas ਕਿਰਪਾ ਕਰਕੇ ਸਾਡੇ ਪਾ ਦੇਵੇਗਾ DataFrame ਦੇ ਰੂਪ ਵਿੱਚ CSV-ਲਾਈਨਾਂ।
ਆਉ ਇੱਕ ਹੁੱਕ ਨਾਲ ਸਾਡੇ ਮਨਪਸੰਦ ਵਰਟਿਕਾ ਨਾਲ ਇੱਕ ਕੁਨੈਕਸ਼ਨ ਖੋਲ੍ਹੀਏ।
ਅਤੇ ਹੁਣ ਮਦਦ ਨਾਲ copy() ਸਾਡਾ ਡੇਟਾ ਸਿੱਧਾ ਵਰਟਿਕਾ ਨੂੰ ਭੇਜੋ!

ਅਸੀਂ ਡਰਾਈਵਰ ਤੋਂ ਲਵਾਂਗੇ ਕਿ ਕਿੰਨੀਆਂ ਲਾਈਨਾਂ ਭਰੀਆਂ ਗਈਆਂ ਸਨ, ਅਤੇ ਸੈਸ਼ਨ ਮੈਨੇਜਰ ਨੂੰ ਦੱਸਾਂਗੇ ਕਿ ਸਭ ਕੁਝ ਠੀਕ ਹੈ:

session.loaded_rows = cursor.rowcount
session.successful = True

ਇਹ ਸਭ ਹੈ.

ਵਿਕਰੀ 'ਤੇ, ਅਸੀਂ ਨਿਸ਼ਾਨਾ ਪਲੇਟ ਹੱਥੀਂ ਬਣਾਉਂਦੇ ਹਾਂ। ਇੱਥੇ ਮੈਂ ਆਪਣੇ ਆਪ ਨੂੰ ਇੱਕ ਛੋਟੀ ਮਸ਼ੀਨ ਦੀ ਇਜਾਜ਼ਤ ਦਿੱਤੀ:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

ਮੈਂ ਵਰਤ ਰਿਹਾ ਹਾਂ VerticaOperator() ਮੈਂ ਇੱਕ ਡੇਟਾਬੇਸ ਸਕੀਮਾ ਅਤੇ ਇੱਕ ਸਾਰਣੀ ਬਣਾਉਂਦਾ ਹਾਂ (ਜੇ ਉਹ ਪਹਿਲਾਂ ਤੋਂ ਮੌਜੂਦ ਨਹੀਂ ਹਨ, ਬੇਸ਼ਕ)। ਮੁੱਖ ਗੱਲ ਇਹ ਹੈ ਕਿ ਨਿਰਭਰਤਾ ਨੂੰ ਸਹੀ ਢੰਗ ਨਾਲ ਵਿਵਸਥਿਤ ਕਰਨਾ:

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

ਸੰਖੇਪ

- ਠੀਕ ਹੈ, - ਛੋਟੇ ਚੂਹੇ ਨੇ ਕਿਹਾ, - ਹੁਣ ਇਹ ਨਹੀਂ ਹੈ
ਕੀ ਤੁਹਾਨੂੰ ਯਕੀਨ ਹੈ ਕਿ ਮੈਂ ਜੰਗਲ ਦਾ ਸਭ ਤੋਂ ਭਿਆਨਕ ਜਾਨਵਰ ਹਾਂ?

ਜੂਲੀਆ ਡੋਨਾਲਡਸਨ, ਦ ਗ੍ਰਫਾਲੋ

ਮੈਨੂੰ ਲਗਦਾ ਹੈ ਕਿ ਜੇ ਮੇਰੇ ਸਾਥੀਆਂ ਅਤੇ ਮੇਰੇ ਵਿੱਚ ਇੱਕ ਮੁਕਾਬਲਾ ਸੀ: ਕੌਣ ਸਕ੍ਰੈਚ ਤੋਂ ਇੱਕ ETL ਪ੍ਰਕਿਰਿਆ ਨੂੰ ਜਲਦੀ ਬਣਾਵੇਗਾ ਅਤੇ ਲਾਂਚ ਕਰੇਗਾ: ਉਹ ਆਪਣੇ SSIS ਅਤੇ ਇੱਕ ਮਾਊਸ ਨਾਲ ਅਤੇ ਮੈਂ ਏਅਰਫਲੋ ਨਾਲ ... ਅਤੇ ਫਿਰ ਅਸੀਂ ਰੱਖ-ਰਖਾਅ ਦੀ ਸੌਖ ਦੀ ਤੁਲਨਾ ਵੀ ਕਰਾਂਗੇ ... ਵਾਹ, ਮੈਨੂੰ ਲਗਦਾ ਹੈ ਕਿ ਤੁਸੀਂ ਸਹਿਮਤ ਹੋਵੋਗੇ ਕਿ ਮੈਂ ਉਨ੍ਹਾਂ ਨੂੰ ਹਰ ਮੋਰਚੇ 'ਤੇ ਹਰਾਵਾਂਗਾ!

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

ਇਸਦੀ ਅਸੀਮਿਤ ਵਿਸਤਾਰਯੋਗਤਾ, ਪਲੱਗ-ਇਨਾਂ ਅਤੇ ਸਕੇਲੇਬਿਲਟੀ ਦੀ ਪ੍ਰਵਿਰਤੀ ਦੇ ਰੂਪ ਵਿੱਚ, ਤੁਹਾਨੂੰ ਲਗਭਗ ਕਿਸੇ ਵੀ ਖੇਤਰ ਵਿੱਚ ਏਅਰਫਲੋ ਦੀ ਵਰਤੋਂ ਕਰਨ ਦਾ ਮੌਕਾ ਦਿੰਦੀ ਹੈ: ਇੱਥੋਂ ਤੱਕ ਕਿ ਡੇਟਾ ਇਕੱਠਾ ਕਰਨ, ਤਿਆਰ ਕਰਨ ਅਤੇ ਪ੍ਰੋਸੈਸ ਕਰਨ ਦੇ ਪੂਰੇ ਚੱਕਰ ਵਿੱਚ, ਇੱਥੋਂ ਤੱਕ ਕਿ ਰਾਕੇਟ ਲਾਂਚ ਕਰਨ ਵਿੱਚ ਵੀ (ਮੰਗਲ ਤੱਕ, ਕੋਰਸ).

ਭਾਗ ਫਾਈਨਲ, ਹਵਾਲਾ ਅਤੇ ਜਾਣਕਾਰੀ

ਰੇਕ ਅਸੀਂ ਤੁਹਾਡੇ ਲਈ ਇਕੱਠਾ ਕੀਤਾ ਹੈ

start_date. ਹਾਂ, ਇਹ ਪਹਿਲਾਂ ਹੀ ਇੱਕ ਸਥਾਨਕ ਮੀਮ ਹੈ। ਡੌਗ ਦੀ ਮੁੱਖ ਦਲੀਲ ਰਾਹੀਂ start_date ਸਾਰੇ ਪਾਸ ਸੰਖੇਪ ਵਿੱਚ, ਜੇਕਰ ਤੁਸੀਂ ਵਿੱਚ ਨਿਸ਼ਚਿਤ ਕਰਦੇ ਹੋ start_date ਮੌਜੂਦਾ ਮਿਤੀ, ਅਤੇ schedule_interval - ਇੱਕ ਦਿਨ, ਫਿਰ ਡੀਏਜੀ ਕੱਲ੍ਹ ਤੋਂ ਪਹਿਲਾਂ ਨਹੀਂ ਸ਼ੁਰੂ ਹੋਵੇਗਾ।
```
start_date = datetime(2020, 7, 7, 0, 1, 2)
```
ਅਤੇ ਕੋਈ ਹੋਰ ਸਮੱਸਿਆ ਨਹੀਂ.

ਇਸਦੇ ਨਾਲ ਜੁੜੀ ਇੱਕ ਹੋਰ ਰਨਟਾਈਮ ਗਲਤੀ ਹੈ: Task is missing the start_date parameter, ਜੋ ਅਕਸਰ ਇਹ ਦਰਸਾਉਂਦਾ ਹੈ ਕਿ ਤੁਸੀਂ ਡੈਗ ਓਪਰੇਟਰ ਨਾਲ ਬੰਨ੍ਹਣਾ ਭੁੱਲ ਗਏ ਹੋ।
ਸਾਰੇ ਇੱਕ ਮਸ਼ੀਨ 'ਤੇ. ਹਾਂ, ਅਤੇ ਬੇਸ (ਏਅਰਫਲੋ ਖੁਦ ਅਤੇ ਸਾਡੀ ਕੋਟਿੰਗ), ਅਤੇ ਇੱਕ ਵੈਬ ਸਰਵਰ, ਅਤੇ ਇੱਕ ਸ਼ਡਿਊਲਰ, ਅਤੇ ਵਰਕਰ। ਅਤੇ ਇਹ ਵੀ ਕੰਮ ਕੀਤਾ. ਪਰ ਸਮੇਂ ਦੇ ਨਾਲ, ਸੇਵਾਵਾਂ ਲਈ ਕਾਰਜਾਂ ਦੀ ਗਿਣਤੀ ਵਧਦੀ ਗਈ, ਅਤੇ ਜਦੋਂ PostgreSQL ਨੇ 20 ms ਦੀ ਬਜਾਏ 5 s ਵਿੱਚ ਸੂਚਕਾਂਕ ਦਾ ਜਵਾਬ ਦੇਣਾ ਸ਼ੁਰੂ ਕੀਤਾ, ਅਸੀਂ ਇਸਨੂੰ ਲੈ ਲਿਆ ਅਤੇ ਇਸਨੂੰ ਦੂਰ ਕਰ ਦਿੱਤਾ।
ਲੋਕਲ ਐਗਜ਼ੀਕਿਊਟਰ। ਹਾਂ, ਅਸੀਂ ਅਜੇ ਵੀ ਇਸ 'ਤੇ ਬੈਠੇ ਹਾਂ, ਅਤੇ ਅਸੀਂ ਪਹਿਲਾਂ ਹੀ ਅਥਾਹ ਕੁੰਡ ਦੇ ਕਿਨਾਰੇ ਆ ਗਏ ਹਾਂ. LocalExecutor ਹੁਣ ਤੱਕ ਸਾਡੇ ਲਈ ਕਾਫੀ ਰਿਹਾ ਹੈ, ਪਰ ਹੁਣ ਸਮਾਂ ਆ ਗਿਆ ਹੈ ਕਿ ਘੱਟੋ-ਘੱਟ ਇੱਕ ਵਰਕਰ ਨਾਲ ਵਿਸਤਾਰ ਕੀਤਾ ਜਾਵੇ, ਅਤੇ ਸਾਨੂੰ CeleryExecutor ਵਿੱਚ ਜਾਣ ਲਈ ਸਖ਼ਤ ਮਿਹਨਤ ਕਰਨੀ ਪਵੇਗੀ। ਅਤੇ ਇਸ ਤੱਥ ਦੇ ਮੱਦੇਨਜ਼ਰ ਕਿ ਤੁਸੀਂ ਇੱਕ ਮਸ਼ੀਨ 'ਤੇ ਇਸ ਨਾਲ ਕੰਮ ਕਰ ਸਕਦੇ ਹੋ, ਤੁਹਾਨੂੰ ਸਰਵਰ 'ਤੇ ਵੀ ਸੈਲਰੀ ਦੀ ਵਰਤੋਂ ਕਰਨ ਤੋਂ ਕੁਝ ਨਹੀਂ ਰੋਕਦਾ, ਜੋ ਕਿ "ਬੇਸ਼ਕ, ਕਦੇ ਵੀ ਉਤਪਾਦਨ ਵਿੱਚ ਨਹੀਂ ਜਾਵੇਗਾ, ਇਮਾਨਦਾਰੀ ਨਾਲ!"
ਗੈਰ-ਵਰਤੋਂ ਬਿਲਟ-ਇਨ ਟੂਲ:
- ਕੁਨੈਕਸ਼ਨ ਸੇਵਾ ਪ੍ਰਮਾਣ ਪੱਤਰਾਂ ਨੂੰ ਸਟੋਰ ਕਰਨ ਲਈ,
- SLA ਮਿਸ ਸਮੇਂ ਸਿਰ ਕੰਮ ਨਾ ਕਰਨ ਵਾਲੇ ਕੰਮਾਂ ਦਾ ਜਵਾਬ ਦੇਣ ਲਈ,
- xcom ਮੈਟਾਡੇਟਾ ਐਕਸਚੇਂਜ ਲਈ (ਮੈਂ ਕਿਹਾ ਮੈਟਾਡਾਟਾ!) ਡੈਗ ਕੰਮਾਂ ਦੇ ਵਿਚਕਾਰ।
ਮੇਲ ਦੁਰਵਿਵਹਾਰ। ਖੈਰ, ਮੈਂ ਕੀ ਕਹਿ ਸਕਦਾ ਹਾਂ? ਡਿੱਗੇ ਹੋਏ ਕੰਮਾਂ ਦੇ ਸਾਰੇ ਦੁਹਰਾਓ ਲਈ ਅਲਰਟ ਸਥਾਪਤ ਕੀਤੇ ਗਏ ਸਨ। ਹੁਣ ਮੇਰਾ ਕੰਮ Gmail ਵਿੱਚ ਏਅਰਫਲੋ ਤੋਂ >90k ਈਮੇਲਾਂ ਹਨ, ਅਤੇ ਵੈਬ ਮੇਲ ਮਜ਼ਲ ਇੱਕ ਸਮੇਂ ਵਿੱਚ 100 ਤੋਂ ਵੱਧ ਨੂੰ ਚੁੱਕਣ ਅਤੇ ਮਿਟਾਉਣ ਤੋਂ ਇਨਕਾਰ ਕਰਦਾ ਹੈ।

ਹੋਰ ਨੁਕਸਾਨ: ਅਪਾਚੇ ਏਅਰਫਲੋ ਪਿਟਫੇਲਜ਼

ਹੋਰ ਆਟੋਮੇਸ਼ਨ ਟੂਲ

ਸਾਡੇ ਹੱਥਾਂ ਨਾਲ ਨਹੀਂ, ਸਾਡੇ ਸਿਰਾਂ ਨਾਲ ਹੋਰ ਵੀ ਕੰਮ ਕਰਨ ਲਈ, ਏਅਰਫਲੋ ਨੇ ਸਾਡੇ ਲਈ ਇਹ ਤਿਆਰ ਕੀਤਾ ਹੈ:

REST API - ਉਸ ਕੋਲ ਅਜੇ ਵੀ ਪ੍ਰਯੋਗਾਤਮਕ ਦਾ ਦਰਜਾ ਹੈ, ਜੋ ਉਸਨੂੰ ਕੰਮ ਕਰਨ ਤੋਂ ਨਹੀਂ ਰੋਕਦਾ। ਇਸ ਦੇ ਨਾਲ, ਤੁਸੀਂ ਨਾ ਸਿਰਫ਼ ਡੈਗ ਅਤੇ ਕੰਮਾਂ ਬਾਰੇ ਜਾਣਕਾਰੀ ਪ੍ਰਾਪਤ ਕਰ ਸਕਦੇ ਹੋ, ਸਗੋਂ ਇੱਕ ਡੈਗ ਨੂੰ ਰੋਕ/ਸ਼ੁਰੂ ਕਰ ਸਕਦੇ ਹੋ, ਇੱਕ ਡੀਏਜੀ ਰਨ ਜਾਂ ਇੱਕ ਪੂਲ ਬਣਾ ਸਕਦੇ ਹੋ।
CLI - ਕਮਾਂਡ ਲਾਈਨ ਰਾਹੀਂ ਬਹੁਤ ਸਾਰੇ ਟੂਲ ਉਪਲਬਧ ਹਨ ਜੋ WebUI ਰਾਹੀਂ ਵਰਤਣ ਲਈ ਅਸੁਵਿਧਾਜਨਕ ਨਹੀਂ ਹਨ, ਪਰ ਆਮ ਤੌਰ 'ਤੇ ਗੈਰਹਾਜ਼ਰ ਹਨ। ਉਦਾਹਰਣ ਲਈ:
- backfill ਕਾਰਜ ਉਦਾਹਰਨਾਂ ਨੂੰ ਮੁੜ ਚਾਲੂ ਕਰਨ ਦੀ ਲੋੜ ਹੈ।
  ਉਦਾਹਰਣ ਵਜੋਂ, ਵਿਸ਼ਲੇਸ਼ਕ ਆਏ ਅਤੇ ਕਿਹਾ: “ਅਤੇ ਤੁਸੀਂ, ਕਾਮਰੇਡ, 1 ਤੋਂ 13 ਜਨਵਰੀ ਤੱਕ ਦੇ ਅੰਕੜਿਆਂ ਵਿੱਚ ਬਕਵਾਸ ਹੈ! ਇਸਨੂੰ ਠੀਕ ਕਰੋ, ਇਸਨੂੰ ਠੀਕ ਕਰੋ, ਇਸਨੂੰ ਠੀਕ ਕਰੋ, ਇਸਨੂੰ ਠੀਕ ਕਰੋ!" ਅਤੇ ਤੁਸੀਂ ਅਜਿਹੇ ਹੌਬ ਹੋ:
```
airflow backfill -s '2020-01-01' -e '2020-01-13' orders
```
- ਅਧਾਰ ਸੇਵਾ: initdb, resetdb, upgradedb, checkdb.
- run, ਜੋ ਤੁਹਾਨੂੰ ਇੱਕ ਉਦਾਹਰਨ ਕਾਰਜ ਨੂੰ ਚਲਾਉਣ, ਅਤੇ ਸਾਰੀਆਂ ਨਿਰਭਰਤਾਵਾਂ 'ਤੇ ਵੀ ਸਕੋਰ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦਾ ਹੈ। ਇਸ ਤੋਂ ਇਲਾਵਾ, ਤੁਸੀਂ ਇਸਨੂੰ ਦੁਆਰਾ ਚਲਾ ਸਕਦੇ ਹੋ LocalExecutor, ਭਾਵੇਂ ਤੁਹਾਡੇ ਕੋਲ ਸੈਲਰੀ ਕਲੱਸਟਰ ਹੈ।
- ਪਰੈਟੀ ਬਹੁਤ ਕੁਝ ਉਹੀ ਕਰਦਾ ਹੈ test, ਸਿਰਫ਼ ਆਧਾਰਾਂ ਵਿੱਚ ਵੀ ਕੁਝ ਨਹੀਂ ਲਿਖਦਾ।
- connections ਸ਼ੈੱਲ ਤੋਂ ਕੁਨੈਕਸ਼ਨ ਬਣਾਉਣ ਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦਾ ਹੈ।
ਪਾਈਥਨ ਏਪੀਆਈ - ਪਰਸਪਰ ਕ੍ਰਿਆ ਕਰਨ ਦਾ ਇੱਕ ਸਖ਼ਤ ਤਰੀਕਾ, ਜੋ ਪਲੱਗਇਨਾਂ ਲਈ ਹੈ, ਅਤੇ ਇਸ ਵਿੱਚ ਛੋਟੇ ਹੱਥਾਂ ਨਾਲ ਝੁਕਣਾ ਨਹੀਂ ਹੈ। ਪਰ ਸਾਨੂੰ ਜਾਣ ਤੋਂ ਕੌਣ ਰੋਕਦਾ ਹੈ /home/airflow/dags, ਰਨ ipython ਅਤੇ ਆਲੇ-ਦੁਆਲੇ ਗੜਬੜ ਸ਼ੁਰੂ? ਤੁਸੀਂ, ਉਦਾਹਰਨ ਲਈ, ਹੇਠਾਂ ਦਿੱਤੇ ਕੋਡ ਨਾਲ ਸਾਰੇ ਕਨੈਕਸ਼ਨਾਂ ਨੂੰ ਨਿਰਯਾਤ ਕਰ ਸਕਦੇ ਹੋ:
```
from airflow import settings
from airflow.models import Connection

fields = 'conn_id conn_type host port schema login password extra'.split()

session = settings.Session()
for conn in session.query(Connection).order_by(Connection.conn_id):
  d = {field: getattr(conn, field) for field in fields}
  print(conn.conn_id, '=', d)
```

ਏਅਰਫਲੋ ਮੈਟਾਡੇਟਾਬੇਸ ਨਾਲ ਜੁੜ ਰਿਹਾ ਹੈ। ਮੈਂ ਇਸ ਨੂੰ ਲਿਖਣ ਦੀ ਸਿਫਾਰਸ਼ ਨਹੀਂ ਕਰਦਾ ਹਾਂ, ਪਰ ਵੱਖ-ਵੱਖ ਖਾਸ ਮੈਟ੍ਰਿਕਸ ਲਈ ਟਾਸਕ ਸਟੇਟਸ ਪ੍ਰਾਪਤ ਕਰਨਾ ਕਿਸੇ ਵੀ API ਦੀ ਵਰਤੋਂ ਕਰਨ ਨਾਲੋਂ ਬਹੁਤ ਤੇਜ਼ ਅਤੇ ਆਸਾਨ ਹੋ ਸਕਦਾ ਹੈ.

ਦੱਸ ਦੇਈਏ ਕਿ ਸਾਡੇ ਸਾਰੇ ਕੰਮ ਬੇਕਾਬੂ ਨਹੀਂ ਹੁੰਦੇ, ਪਰ ਉਹ ਕਈ ਵਾਰ ਡਿੱਗ ਸਕਦੇ ਹਨ, ਅਤੇ ਇਹ ਆਮ ਗੱਲ ਹੈ। ਪਰ ਕੁਝ ਰੁਕਾਵਟਾਂ ਪਹਿਲਾਂ ਹੀ ਸ਼ੱਕੀ ਹਨ, ਅਤੇ ਇਸਦੀ ਜਾਂਚ ਕਰਨੀ ਜ਼ਰੂਰੀ ਹੋਵੇਗੀ।

SQL ਸਾਵਧਾਨ!

WITH last_executions AS (
SELECT
    task_id,
    dag_id,
    execution_date,
    state,
        row_number()
        OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC) AS rn
FROM public.task_instance
WHERE
    execution_date > now() - INTERVAL '2' DAY
),
failed AS (
    SELECT
        task_id,
        dag_id,
        execution_date,
        state,
        CASE WHEN rn = row_number() OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC)
                 THEN TRUE END AS last_fail_seq
    FROM last_executions
    WHERE
        state IN ('failed', 'up_for_retry')
)
SELECT
    task_id,
    dag_id,
    count(last_fail_seq)                       AS unsuccessful,
    count(CASE WHEN last_fail_seq
        AND state = 'failed' THEN 1 END)       AS failed,
    count(CASE WHEN last_fail_seq
        AND state = 'up_for_retry' THEN 1 END) AS up_for_retry
FROM failed
GROUP BY
    task_id,
    dag_id
HAVING
    count(last_fail_seq) > 0

ਹਵਾਲੇ

ਅਤੇ ਬੇਸ਼ੱਕ, ਗੂਗਲ ਦੇ ਜਾਰੀ ਹੋਣ ਤੋਂ ਪਹਿਲੇ ਦਸ ਲਿੰਕ ਮੇਰੇ ਬੁੱਕਮਾਰਕਸ ਤੋਂ ਏਅਰਫਲੋ ਫੋਲਡਰ ਦੀ ਸਮੱਗਰੀ ਹਨ.

ਅਪਾਚੇ ਏਅਰਫਲੋ ਦਸਤਾਵੇਜ਼ - ਬੇਸ਼ੱਕ, ਸਾਨੂੰ ਦਫ਼ਤਰ ਨਾਲ ਸ਼ੁਰੂ ਕਰਨਾ ਚਾਹੀਦਾ ਹੈ. ਦਸਤਾਵੇਜ਼, ਪਰ ਨਿਰਦੇਸ਼ਾਂ ਨੂੰ ਕੌਣ ਪੜ੍ਹਦਾ ਹੈ?
ਵਧੀਆ ਪ੍ਰੈਕਟਿਸ - ਠੀਕ ਹੈ, ਘੱਟੋ-ਘੱਟ ਸਿਰਜਣਹਾਰਾਂ ਦੀਆਂ ਸਿਫ਼ਾਰਸ਼ਾਂ ਪੜ੍ਹੋ।
ਏਅਰਫਲੋ UI - ਬਹੁਤ ਹੀ ਸ਼ੁਰੂਆਤ: ਤਸਵੀਰਾਂ ਵਿੱਚ ਉਪਭੋਗਤਾ ਇੰਟਰਫੇਸ
ਅਪਾਚੇ ਏਅਰਫਲੋ ਦੇ ਮੁੱਖ ਸੰਕਲਪਾਂ ਨੂੰ ਸਮਝਣਾ - ਬੁਨਿਆਦੀ ਸੰਕਲਪਾਂ ਦਾ ਚੰਗੀ ਤਰ੍ਹਾਂ ਵਰਣਨ ਕੀਤਾ ਗਿਆ ਹੈ, ਜੇਕਰ (ਅਚਾਨਕ!) ਤੁਸੀਂ ਮੇਰੇ ਤੋਂ ਕੁਝ ਨਹੀਂ ਸਮਝਿਆ.
ਟਿਆਨਲੋਂਗ ਦਾ ਬਲੌਗ - ਇੱਕ ਏਅਰਫਲੋ ਸਰਵਰ/ਕਲੱਸਟਰ ਕਿਵੇਂ ਬਣਾਉਣਾ ਹੈ ਬਾਰੇ ਇੱਕ ਗਾਈਡ - ਇੱਕ ਏਅਰਫਲੋ ਕਲੱਸਟਰ ਸਥਾਪਤ ਕਰਨ ਲਈ ਇੱਕ ਛੋਟੀ ਗਾਈਡ।
ਲਿਫਟ 'ਤੇ ਅਪਾਚੇ ਏਅਰਫਲੋ ਚੱਲ ਰਿਹਾ ਹੈ - ਲਗਭਗ ਉਹੀ ਦਿਲਚਸਪ ਲੇਖ, ਸ਼ਾਇਦ ਵਧੇਰੇ ਰਸਮੀਵਾਦ, ਅਤੇ ਘੱਟ ਉਦਾਹਰਣਾਂ ਨੂੰ ਛੱਡ ਕੇ।
ਅਪਾਚੇ ਏਅਰਫਲੋ ਸੈਲਰੀ ਵਰਕਰਾਂ 'ਤੇ ਨੌਕਰੀਆਂ ਕਿਵੇਂ ਵੰਡਦਾ ਹੈ - ਸੈਲਰੀ ਦੇ ਨਾਲ ਮਿਲ ਕੇ ਕੰਮ ਕਰਨ ਬਾਰੇ।
DAG ਅਪਾਚੇ ਏਅਰਫਲੋ ਵਿੱਚ ਵਧੀਆ ਅਭਿਆਸ ਲਿਖਣਾ - ਕਾਰਜਾਂ ਦੀ ਸੁਤੰਤਰਤਾ ਬਾਰੇ, ਮਿਤੀ ਦੀ ਬਜਾਏ ਆਈਡੀ ਦੁਆਰਾ ਲੋਡ ਕਰਨਾ, ਪਰਿਵਰਤਨ, ਫਾਈਲ ਬਣਤਰ ਅਤੇ ਹੋਰ ਦਿਲਚਸਪ ਚੀਜ਼ਾਂ.
ਅਪਾਚੇ ਏਅਰਫਲੋ ਵਿੱਚ ਨਿਰਭਰਤਾ ਦਾ ਪ੍ਰਬੰਧਨ ਕਰਨਾ - ਕਾਰਜਾਂ ਦੀ ਨਿਰਭਰਤਾ ਅਤੇ ਟਰਿੱਗਰ ਨਿਯਮ, ਜਿਸਦਾ ਮੈਂ ਸਿਰਫ ਪਾਸ ਕਰਨ ਵਿੱਚ ਜ਼ਿਕਰ ਕੀਤਾ ਹੈ।
ਏਅਰਫਲੋ: ਜਦੋਂ ਤੁਹਾਡਾ DAG ਅਨੁਸੂਚੀ ਤੋਂ ਬਹੁਤ ਪਿੱਛੇ ਹੈ - ਸ਼ਡਿਊਲਰ ਵਿੱਚ ਕੁਝ "ਇਰਾਦੇ ਅਨੁਸਾਰ ਕੰਮ" ਨੂੰ ਕਿਵੇਂ ਦੂਰ ਕਰਨਾ ਹੈ, ਗੁੰਮ ਹੋਏ ਡੇਟਾ ਨੂੰ ਲੋਡ ਕਰੋ ਅਤੇ ਕਾਰਜਾਂ ਨੂੰ ਤਰਜੀਹ ਦਿਓ।
Apache Airflow ਲਈ ਉਪਯੋਗੀ SQL ਸਵਾਲ - ਏਅਰਫਲੋ ਮੈਟਾਡੇਟਾ ਲਈ ਉਪਯੋਗੀ SQL ਸਵਾਲ।
ਅਪਾਚੇ ਏਅਰਫਲੋ ਨਾਲ ਵਰਕਫਲੋ ਵਿਕਸਿਤ ਕਰਨਾ ਸ਼ੁਰੂ ਕਰੋ - ਇੱਕ ਕਸਟਮ ਸੈਂਸਰ ਬਣਾਉਣ ਬਾਰੇ ਇੱਕ ਉਪਯੋਗੀ ਭਾਗ ਹੈ।
Presto ਅਤੇ Airflow ਦੇ ਨਾਲ AWS 'ਤੇ Fetchr ਡਾਟਾ ਸਾਇੰਸ ਇਨਫਰਾ ਬਣਾਉਣਾ - ਡੇਟਾ ਸਾਇੰਸ ਲਈ AWS 'ਤੇ ਇੱਕ ਬੁਨਿਆਦੀ ਢਾਂਚਾ ਬਣਾਉਣ ਬਾਰੇ ਇੱਕ ਦਿਲਚਸਪ ਛੋਟਾ ਨੋਟ।
ਏਅਰਫਲੋ DAGs ਨੂੰ ਡੀਬੱਗ ਕਰਨ ਵੇਲੇ ਜਾਂਚ ਕਰਨ ਲਈ 7 ਆਮ ਗਲਤੀਆਂ - ਆਮ ਗਲਤੀਆਂ (ਜਦੋਂ ਕੋਈ ਵਿਅਕਤੀ ਅਜੇ ਵੀ ਨਿਰਦੇਸ਼ਾਂ ਨੂੰ ਨਹੀਂ ਪੜ੍ਹਦਾ).
ਅਪਾਚੇ ਏਅਰਫਲੋ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਪਾਸਵਰਡ ਸਟੋਰ ਅਤੇ ਐਕਸੈਸ ਕਰੋ - ਮੁਸਕਰਾਓ ਕਿ ਕਿਵੇਂ ਲੋਕ ਪਾਸਵਰਡ ਸਟੋਰ ਕਰਦੇ ਹਨ, ਹਾਲਾਂਕਿ ਤੁਸੀਂ ਸਿਰਫ਼ ਕਨੈਕਸ਼ਨਾਂ ਦੀ ਵਰਤੋਂ ਕਰ ਸਕਦੇ ਹੋ।
ਪਾਈਥਨ ਅਤੇ ਅਪਾਚੇ ਏਅਰਫਲੋ ਦਾ ਜ਼ੈਨ - ਅਪ੍ਰਤੱਖ DAG ਫਾਰਵਰਡਿੰਗ, ਫੰਕਸ਼ਨਾਂ ਵਿੱਚ ਸੰਦਰਭ ਸੁੱਟਣਾ, ਮੁੜ ਨਿਰਭਰਤਾ ਬਾਰੇ, ਅਤੇ ਟਾਸਕ ਲਾਂਚਾਂ ਨੂੰ ਛੱਡਣ ਬਾਰੇ ਵੀ।
ਏਅਰਫਲੋ: ਘੱਟ ਜਾਣੇ-ਪਛਾਣੇ ਸੁਝਾਅ, ਟ੍ਰਿਕਸ ਅਤੇ ਵਧੀਆ ਅਭਿਆਸ - ਵਰਤੋਂ ਬਾਰੇ default arguments и params ਟੈਂਪਲੇਟਾਂ ਵਿੱਚ, ਨਾਲ ਹੀ ਵੇਰੀਏਬਲ ਅਤੇ ਕਨੈਕਸ਼ਨਾਂ ਵਿੱਚ।
ਏਅਰਫਲੋ ਸ਼ਡਿਊਲਰ ਦੀ ਪਰੋਫਾਈਲਿੰਗ - ਇਸ ਬਾਰੇ ਇੱਕ ਕਹਾਣੀ ਕਿ ਕਿਵੇਂ ਯੋਜਨਾਕਾਰ ਏਅਰਫਲੋ 2.0 ਲਈ ਤਿਆਰੀ ਕਰ ਰਿਹਾ ਹੈ।
ਡੌਕਰ-ਕੰਪੋਜ਼ ਵਿੱਚ 3 ਸੈਲਰੀ ਵਰਕਰਾਂ ਨਾਲ ਅਪਾਚੇ ਏਅਰਫਲੋ - ਵਿੱਚ ਸਾਡੇ ਕਲੱਸਟਰ ਨੂੰ ਤੈਨਾਤ ਕਰਨ ਬਾਰੇ ਇੱਕ ਥੋੜ੍ਹਾ ਪੁਰਾਣਾ ਲੇਖ docker-compose.
ਏਅਰਫਲੋ ਸੰਦਰਭ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ 4 ਟੈਂਪਲੇਟਿੰਗ ਟਾਸਕ - ਟੈਂਪਲੇਟਸ ਅਤੇ ਸੰਦਰਭ ਫਾਰਵਰਡਿੰਗ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹੋਏ ਗਤੀਸ਼ੀਲ ਕਾਰਜ।
ਏਅਰਫਲੋ ਵਿੱਚ ਗਲਤੀ ਸੂਚਨਾਵਾਂ - ਮੇਲ ਅਤੇ ਸਲੈਕ ਦੁਆਰਾ ਮਿਆਰੀ ਅਤੇ ਕਸਟਮ ਸੂਚਨਾਵਾਂ।
ਏਅਰਫਲੋ ਵਰਕਸ਼ਾਪ: ਬਿਨਾਂ ਬੈਸਾਖੀਆਂ ਦੇ ਗੁੰਝਲਦਾਰ ਡੀਏਜੀ - ਬ੍ਰਾਂਚਿੰਗ ਟਾਸਕ, ਮੈਕਰੋ ਅਤੇ ਐਕਸਕਾਮ।

ਅਤੇ ਲੇਖ ਵਿੱਚ ਵਰਤੇ ਗਏ ਲਿੰਕ:

ਮੈਕਰੋ ਹਵਾਲਾ - ਟੈਂਪਲੇਟਸ ਵਿੱਚ ਵਰਤਣ ਲਈ ਪਲੇਸਹੋਲਡਰ ਉਪਲਬਧ ਹਨ।
ਆਮ ਨੁਕਸਾਨ - ਏਅਰਫਲੋ - ਡੈਗ ਬਣਾਉਣ ਵੇਲੇ ਆਮ ਗਲਤੀਆਂ।
puckel/docker-airflow: Docker Apache Airflow - docker-compose ਪ੍ਰਯੋਗ, ਡੀਬੱਗਿੰਗ ਅਤੇ ਹੋਰ ਲਈ।
python-telegram-bot/python-telegram-bot: ਅਸੀਂ ਤੁਹਾਨੂੰ ਇੱਕ ਰੈਪਰ ਬਣਾਇਆ ਹੈ ਜਿਸ ਤੋਂ ਤੁਸੀਂ ਇਨਕਾਰ ਨਹੀਂ ਕਰ ਸਕਦੇ - ਟੈਲੀਗ੍ਰਾਮ REST API ਲਈ ਪਾਈਥਨ ਰੈਪਰ।

ਸਰੋਤ: www.habr.com

ਅਪਾਚੇ ਏਅਰਫਲੋ: ਈਟੀਐਲ ਨੂੰ ਆਸਾਨ ਬਣਾਉਣਾ