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

ਹੈਲੋ, ਹੈਬਰ! ਇਸ ਲੇਖ ਵਿੱਚ ਮੈਂ ਬੈਚ ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਪ੍ਰਕਿਰਿਆਵਾਂ ਨੂੰ ਵਿਕਸਤ ਕਰਨ ਲਈ ਇੱਕ ਵਧੀਆ ਟੂਲ ਬਾਰੇ ਗੱਲ ਕਰਨਾ ਚਾਹੁੰਦਾ ਹਾਂ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ ਕਾਰਪੋਰੇਟ DWH ਜਾਂ ਤੁਹਾਡੇ ਡੇਟਾਲੇਕ ਦੇ ਬੁਨਿਆਦੀ ਢਾਂਚੇ ਵਿੱਚ. ਅਸੀਂ ਅਪਾਚੇ ਏਅਰਫਲੋ (ਇਸ ਤੋਂ ਬਾਅਦ ਏਅਰਫਲੋ ਕਿਹਾ ਜਾਂਦਾ ਹੈ) ਬਾਰੇ ਗੱਲ ਕਰਾਂਗੇ। ਇਹ ਹੈਬਰੇ 'ਤੇ ਧਿਆਨ ਦੇਣ ਤੋਂ ਵਾਂਝਾ ਹੈ, ਅਤੇ ਮੁੱਖ ਹਿੱਸੇ ਵਿੱਚ ਮੈਂ ਤੁਹਾਨੂੰ ਯਕੀਨ ਦਿਵਾਉਣ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਾਂਗਾ ਕਿ ਤੁਹਾਡੀਆਂ ETL/ELT ਪ੍ਰਕਿਰਿਆਵਾਂ ਲਈ ਇੱਕ ਸ਼ਡਿਊਲਰ ਦੀ ਚੋਣ ਕਰਦੇ ਸਮੇਂ ਘੱਟੋ-ਘੱਟ ਏਅਰਫਲੋ ਦੇਖਣ ਦੇ ਯੋਗ ਹੈ।

ਪਹਿਲਾਂ, ਜਦੋਂ ਮੈਂ ਟਿੰਕੋਫ ਬੈਂਕ ਵਿੱਚ ਕੰਮ ਕੀਤਾ ਸੀ ਤਾਂ ਮੈਂ DWH ਦੇ ਵਿਸ਼ੇ 'ਤੇ ਲੇਖਾਂ ਦੀ ਇੱਕ ਲੜੀ ਲਿਖੀ ਸੀ। ਹੁਣ ਮੈਂ Mail.Ru ਗਰੁੱਪ ਟੀਮ ਦਾ ਹਿੱਸਾ ਬਣ ਗਿਆ ਹਾਂ ਅਤੇ ਗੇਮਿੰਗ ਖੇਤਰ ਵਿੱਚ ਡਾਟਾ ਵਿਸ਼ਲੇਸ਼ਣ ਲਈ ਇੱਕ ਪਲੇਟਫਾਰਮ ਤਿਆਰ ਕਰ ਰਿਹਾ/ਰਹੀ ਹਾਂ। ਅਸਲ ਵਿੱਚ, ਜਿਵੇਂ ਹੀ ਖਬਰਾਂ ਅਤੇ ਦਿਲਚਸਪ ਹੱਲ ਦਿਖਾਈ ਦਿੰਦੇ ਹਨ, ਮੈਂ ਅਤੇ ਮੇਰੀ ਟੀਮ ਇੱਥੇ ਡੇਟਾ ਵਿਸ਼ਲੇਸ਼ਣ ਲਈ ਸਾਡੇ ਪਲੇਟਫਾਰਮ ਬਾਰੇ ਗੱਲ ਕਰਾਂਗੇ।

ਪ੍ਰੋਲੋਗ

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

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

ਡੀ.ਏ.ਜੀ.

ਇੱਕ ਡੀਏਜੀ ਤੁਹਾਡੇ ਕਾਰਜਾਂ ਦਾ ਕੁਝ ਅਰਥਪੂਰਨ ਸਬੰਧ ਹੈ ਜੋ ਤੁਸੀਂ ਇੱਕ ਖਾਸ ਅਨੁਸੂਚੀ ਦੇ ਅਨੁਸਾਰ ਇੱਕ ਸਖਤੀ ਨਾਲ ਪਰਿਭਾਸ਼ਿਤ ਕ੍ਰਮ ਵਿੱਚ ਪੂਰਾ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹੋ। ਏਅਰਫਲੋ DAGs ਅਤੇ ਹੋਰ ਸੰਸਥਾਵਾਂ ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਇੱਕ ਸੁਵਿਧਾਜਨਕ ਵੈੱਬ ਇੰਟਰਫੇਸ ਪ੍ਰਦਾਨ ਕਰਦਾ ਹੈ:

DAG ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦੇ ਸਕਦਾ ਹੈ:

ਡਿਵੈਲਪਰ, ਜਦੋਂ ਇੱਕ ਡੀਏਜੀ ਡਿਜ਼ਾਈਨ ਕਰਦਾ ਹੈ, ਓਪਰੇਟਰਾਂ ਦਾ ਇੱਕ ਸੈੱਟ ਰੱਖਦਾ ਹੈ ਜਿਸ ਉੱਤੇ ਡੀਏਜੀ ਦੇ ਅੰਦਰ ਕੰਮ ਬਣਾਏ ਜਾਣਗੇ। ਇੱਥੇ ਅਸੀਂ ਇੱਕ ਹੋਰ ਮਹੱਤਵਪੂਰਨ ਸੰਸਥਾ ਵੱਲ ਆਉਂਦੇ ਹਾਂ: ਏਅਰਫਲੋ ਆਪਰੇਟਰ।

ਓਪਰੇਟਰ

ਇੱਕ ਆਪਰੇਟਰ ਇੱਕ ਅਜਿਹੀ ਇਕਾਈ ਹੁੰਦੀ ਹੈ ਜਿਸ ਦੇ ਅਧਾਰ 'ਤੇ ਨੌਕਰੀ ਦੀਆਂ ਉਦਾਹਰਣਾਂ ਬਣਾਈਆਂ ਜਾਂਦੀਆਂ ਹਨ, ਜੋ ਦੱਸਦੀ ਹੈ ਕਿ ਨੌਕਰੀ ਦੀ ਉਦਾਹਰਨ ਦੇ ਅਮਲ ਦੌਰਾਨ ਕੀ ਹੋਵੇਗਾ। GitHub ਤੋਂ ਏਅਰਫਲੋ ਰਿਲੀਜ਼ ਹੁੰਦੀ ਹੈ ਪਹਿਲਾਂ ਹੀ ਵਰਤਣ ਲਈ ਤਿਆਰ ਓਪਰੇਟਰਾਂ ਦਾ ਇੱਕ ਸੈੱਟ ਰੱਖਦਾ ਹੈ। ਉਦਾਹਰਨਾਂ:

• BashOperator - ਇੱਕ bash ਕਮਾਂਡ ਚਲਾਉਣ ਲਈ ਆਪਰੇਟਰ।
• PythonOperator - Python ਕੋਡ ਨੂੰ ਕਾਲ ਕਰਨ ਲਈ ਆਪਰੇਟਰ।
• EmailOperator — ਈਮੇਲ ਭੇਜਣ ਲਈ ਆਪਰੇਟਰ।
• HTTPOperator - http ਬੇਨਤੀਆਂ ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਆਪਰੇਟਰ।
• SqlOperator - SQL ਕੋਡ ਨੂੰ ਚਲਾਉਣ ਲਈ ਆਪਰੇਟਰ।
• ਸੈਂਸਰ ਇੱਕ ਇਵੈਂਟ ਦੀ ਉਡੀਕ ਕਰਨ ਲਈ ਇੱਕ ਓਪਰੇਟਰ ਹੈ (ਲੋੜੀਂਦੇ ਸਮੇਂ ਦਾ ਆਗਮਨ, ਲੋੜੀਂਦੀ ਫਾਈਲ ਦੀ ਦਿੱਖ, ਡੇਟਾਬੇਸ ਵਿੱਚ ਇੱਕ ਲਾਈਨ, API ਤੋਂ ਜਵਾਬ, ਆਦਿ, ਆਦਿ)।

ਇੱਥੇ ਹੋਰ ਖਾਸ ਓਪਰੇਟਰ ਹਨ: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator।

ਤੁਸੀਂ ਆਪਣੀਆਂ ਵਿਸ਼ੇਸ਼ਤਾਵਾਂ ਦੇ ਆਧਾਰ 'ਤੇ ਆਪਰੇਟਰ ਵੀ ਵਿਕਸਤ ਕਰ ਸਕਦੇ ਹੋ ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਆਪਣੇ ਪ੍ਰੋਜੈਕਟ ਵਿੱਚ ਵਰਤ ਸਕਦੇ ਹੋ। ਉਦਾਹਰਨ ਲਈ, ਅਸੀਂ MongoDBToHiveViaHdfsTransfer, MongoDB ਤੋਂ Hive ਵਿੱਚ ਦਸਤਾਵੇਜ਼ ਨਿਰਯਾਤ ਕਰਨ ਲਈ ਇੱਕ ਆਪਰੇਟਰ, ਅਤੇ ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਕਈ ਓਪਰੇਟਰ ਬਣਾਏ ਹਨ। ਕਲਿਕਹਾouseਸ: CHLoadFromHiveOperator ਅਤੇ CHTableLoaderOperator। ਜ਼ਰੂਰੀ ਤੌਰ 'ਤੇ, ਜਿਵੇਂ ਹੀ ਇੱਕ ਪ੍ਰੋਜੈਕਟ ਨੇ ਬੁਨਿਆਦੀ ਸਟੇਟਮੈਂਟਾਂ 'ਤੇ ਬਣੇ ਕੋਡ ਦੀ ਅਕਸਰ ਵਰਤੋਂ ਕੀਤੀ ਹੈ, ਤੁਸੀਂ ਇਸਨੂੰ ਇੱਕ ਨਵੇਂ ਬਿਆਨ ਵਿੱਚ ਬਣਾਉਣ ਬਾਰੇ ਸੋਚ ਸਕਦੇ ਹੋ। ਇਹ ਹੋਰ ਵਿਕਾਸ ਨੂੰ ਸਰਲ ਬਣਾ ਦੇਵੇਗਾ, ਅਤੇ ਤੁਸੀਂ ਪ੍ਰੋਜੈਕਟ ਵਿੱਚ ਆਪਰੇਟਰਾਂ ਦੀ ਆਪਣੀ ਲਾਇਬ੍ਰੇਰੀ ਦਾ ਵਿਸਤਾਰ ਕਰੋਗੇ।

ਅੱਗੇ, ਕਾਰਜਾਂ ਦੀਆਂ ਇਹਨਾਂ ਸਾਰੀਆਂ ਉਦਾਹਰਣਾਂ ਨੂੰ ਲਾਗੂ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੈ, ਅਤੇ ਹੁਣ ਅਸੀਂ ਸ਼ਡਿਊਲਰ ਬਾਰੇ ਗੱਲ ਕਰਾਂਗੇ.

ਸ਼ਡਿਊਲਰ

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

ਹਰੇਕ ਪੂਲ ਵਿੱਚ ਸਲਾਟਾਂ ਦੀ ਗਿਣਤੀ ਦੀ ਇੱਕ ਸੀਮਾ ਹੁੰਦੀ ਹੈ। DAG ਬਣਾਉਂਦੇ ਸਮੇਂ, ਇਸਨੂੰ ਇੱਕ ਪੂਲ ਦਿੱਤਾ ਜਾਂਦਾ ਹੈ:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

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

1. ਡੀਏਜੀ ਵਿੱਚ ਪਿਛਲੇ ਕੰਮ ਪੂਰੇ ਹੋ ਚੁੱਕੇ ਹਨ, ਇੱਕ ਨਵਾਂ ਕਤਾਰਬੱਧ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ।
2. ਕਤਾਰ ਨੂੰ ਕਾਰਜਾਂ ਦੀ ਤਰਜੀਹ ਦੇ ਅਧਾਰ ਤੇ ਕ੍ਰਮਬੱਧ ਕੀਤਾ ਜਾਂਦਾ ਹੈ (ਪਹਿਲਤਾਵਾਂ ਨੂੰ ਵੀ ਨਿਯੰਤਰਿਤ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ), ਅਤੇ ਜੇ ਪੂਲ ਵਿੱਚ ਇੱਕ ਮੁਫਤ ਸਲਾਟ ਹੈ, ਤਾਂ ਕਾਰਜ ਨੂੰ ਕਾਰਜ ਵਿੱਚ ਲਿਆ ਜਾ ਸਕਦਾ ਹੈ।
3. ਜੇ ਕੋਈ ਮੁਫਤ ਵਰਕਰ ਸੈਲਰੀ ਹੈ, ਤਾਂ ਕੰਮ ਉਸ ਨੂੰ ਭੇਜਿਆ ਜਾਂਦਾ ਹੈ; ਉਹ ਕੰਮ ਜੋ ਤੁਸੀਂ ਸਮੱਸਿਆ ਵਿੱਚ ਪ੍ਰੋਗਰਾਮ ਕੀਤਾ ਹੈ, ਇੱਕ ਜਾਂ ਦੂਜੇ ਆਪਰੇਟਰ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਸ਼ੁਰੂ ਹੁੰਦਾ ਹੈ।

ਕਾਫ਼ੀ ਸਧਾਰਨ.

ਸ਼ਡਿਊਲਰ ਸਾਰੇ DAGs ਅਤੇ DAGs ਦੇ ਅੰਦਰ ਸਾਰੇ ਕਾਰਜਾਂ ਦੇ ਸੈੱਟ 'ਤੇ ਚੱਲਦਾ ਹੈ।

DAG ਨਾਲ ਕੰਮ ਸ਼ੁਰੂ ਕਰਨ ਲਈ ਸ਼ਡਿਊਲਰ ਲਈ, DAG ਨੂੰ ਇੱਕ ਸਮਾਂ-ਸਾਰਣੀ ਸੈੱਟ ਕਰਨ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

ਇੱਥੇ ਤਿਆਰ ਪ੍ਰੀਸੈਟਾਂ ਦਾ ਇੱਕ ਸਮੂਹ ਹੈ: @once, @hourly, @daily, @weekly, @monthly, @yearly.

ਤੁਸੀਂ ਕ੍ਰੋਨ ਸਮੀਕਰਨ ਵੀ ਵਰਤ ਸਕਦੇ ਹੋ:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

ਐਗਜ਼ੀਕਿਊਸ਼ਨ ਦੀ ਮਿਤੀ

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

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

ਕਾਰਜ ਪੈਦਾ ਕਰਨਾ

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

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

ਡੀਏਜੀ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਦਾ ਹੈ:

ਇਸ ਸਥਿਤੀ ਵਿੱਚ, ਤੁਸੀਂ ਸਿਰਫ਼ ਸੈਟਿੰਗਾਂ ਨੂੰ ਵਿਵਸਥਿਤ ਕਰਕੇ ਅਤੇ DAG ਨੂੰ ਅੱਪਡੇਟ ਕਰਕੇ ਇੱਕ ਸ਼ਾਰਡ ਨੂੰ ਜੋੜ ਜਾਂ ਹਟਾ ਸਕਦੇ ਹੋ। ਆਰਾਮਦਾਇਕ!

ਤੁਸੀਂ ਵਧੇਰੇ ਗੁੰਝਲਦਾਰ ਕੋਡ ਜਨਰੇਸ਼ਨ ਦੀ ਵਰਤੋਂ ਵੀ ਕਰ ਸਕਦੇ ਹੋ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ ਡੇਟਾਬੇਸ ਦੇ ਰੂਪ ਵਿੱਚ ਸਰੋਤਾਂ ਨਾਲ ਕੰਮ ਕਰੋ ਜਾਂ ਇੱਕ ਸਾਰਣੀ ਬਣਤਰ ਦਾ ਵਰਣਨ ਕਰੋ, ਇੱਕ ਸਾਰਣੀ ਨਾਲ ਕੰਮ ਕਰਨ ਲਈ ਇੱਕ ਐਲਗੋਰਿਦਮ, ਅਤੇ, DWH ਬੁਨਿਆਦੀ ਢਾਂਚੇ ਦੀਆਂ ਵਿਸ਼ੇਸ਼ਤਾਵਾਂ ਨੂੰ ਧਿਆਨ ਵਿੱਚ ਰੱਖਦੇ ਹੋਏ, ਇੱਕ ਪ੍ਰਕਿਰਿਆ ਤਿਆਰ ਕਰੋ ਤੁਹਾਡੀ ਸਟੋਰੇਜ ਵਿੱਚ N ਟੇਬਲ ਲੋਡ ਕਰਨ ਲਈ। ਜਾਂ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ API ਦੇ ਨਾਲ ਕੰਮ ਕਰਨਾ ਜੋ ਇੱਕ ਸੂਚੀ ਦੇ ਰੂਪ ਵਿੱਚ ਇੱਕ ਪੈਰਾਮੀਟਰ ਨਾਲ ਕੰਮ ਕਰਨ ਦਾ ਸਮਰਥਨ ਨਹੀਂ ਕਰਦਾ ਹੈ, ਤੁਸੀਂ ਇਸ ਸੂਚੀ ਤੋਂ ਇੱਕ DAG ਵਿੱਚ N ਕਾਰਜ ਤਿਆਰ ਕਰ ਸਕਦੇ ਹੋ, API ਵਿੱਚ ਬੇਨਤੀਆਂ ਦੀ ਸਮਾਨਤਾ ਨੂੰ ਇੱਕ ਪੂਲ ਤੱਕ ਸੀਮਤ ਕਰ ਸਕਦੇ ਹੋ, ਅਤੇ ਸਕ੍ਰੈਪ ਕਰ ਸਕਦੇ ਹੋ। API ਤੋਂ ਲੋੜੀਂਦਾ ਡੇਟਾ। ਲਚਕਦਾਰ!

ਰਿਪੋਜ਼ਟਰੀ

ਏਅਰਫਲੋ ਦਾ ਆਪਣਾ ਬੈਕਐਂਡ ਰਿਪੋਜ਼ਟਰੀ ਹੈ, ਇੱਕ ਡੇਟਾਬੇਸ (MySQL ਜਾਂ Postgres ਹੋ ਸਕਦਾ ਹੈ, ਸਾਡੇ ਕੋਲ Postgres ਹੈ), ਜੋ ਕਿ ਕੰਮ ਦੀਆਂ ਸਥਿਤੀਆਂ, DAGs, ਕਨੈਕਸ਼ਨ ਸੈਟਿੰਗਾਂ, ਗਲੋਬਲ ਵੇਰੀਏਬਲ, ਆਦਿ, ਆਦਿ ਨੂੰ ਸਟੋਰ ਕਰਦਾ ਹੈ। ਇੱਥੇ ਮੈਂ ਇਹ ਕਹਿਣਾ ਚਾਹਾਂਗਾ ਕਿ ਏਅਰਫਲੋ ਵਿੱਚ ਰਿਪੋਜ਼ਟਰੀ ਬਹੁਤ ਸਰਲ ਹੈ (ਲਗਭਗ 20 ਟੇਬਲ) ਅਤੇ ਸੁਵਿਧਾਜਨਕ ਹੈ ਜੇਕਰ ਤੁਸੀਂ ਇਸ ਦੇ ਸਿਖਰ 'ਤੇ ਆਪਣੀ ਕੋਈ ਪ੍ਰਕਿਰਿਆ ਬਣਾਉਣਾ ਚਾਹੁੰਦੇ ਹੋ। ਮੈਨੂੰ ਇਨਫਾਰਮੈਟਿਕਾ ਰਿਪੋਜ਼ਟਰੀ ਵਿੱਚ 100500 ਟੇਬਲ ਯਾਦ ਹਨ, ਜਿਨ੍ਹਾਂ ਨੂੰ ਇੱਕ ਸਵਾਲ ਬਣਾਉਣ ਦੇ ਤਰੀਕੇ ਨੂੰ ਸਮਝਣ ਤੋਂ ਪਹਿਲਾਂ ਲੰਬੇ ਸਮੇਂ ਲਈ ਅਧਿਐਨ ਕਰਨਾ ਪੈਂਦਾ ਸੀ।

ਨਿਗਰਾਨੀ

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

ਇਹ ਖੁਦ ਏਅਰਫਲੋ ਦਾ ਵੈੱਬ ਇੰਟਰਫੇਸ ਵੀ ਹੋ ਸਕਦਾ ਹੈ:

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

ਸਾਨੂੰ ਟੈਲੀਗ੍ਰਾਮ (ਜੇ ਲੋੜ ਹੋਵੇ) ਦੁਆਰਾ ਇੱਕ ਤੁਰੰਤ ਜਵਾਬ ਪ੍ਰਾਪਤ ਹੁੰਦਾ ਹੈ, ਅਤੇ Zeppelin ਦੁਆਰਾ ਸਾਨੂੰ Airflow ਵਿੱਚ ਕਾਰਜਾਂ ਦੀ ਇੱਕ ਸਮੁੱਚੀ ਤਸਵੀਰ ਪ੍ਰਾਪਤ ਹੁੰਦੀ ਹੈ।

ਕੁੱਲ

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

ਹੁਣ ਸਾਡੇ ਕੋਲ ਏਅਰਫਲੋ ਰੋਜ਼ਾਨਾ ਕੰਮ ਕਰ ਰਿਹਾ ਹੈ ਲਗਭਗ 6,5 ਹਜ਼ਾਰ ਕੰਮ. ਉਹ ਚਰਿੱਤਰ ਵਿੱਚ ਕਾਫ਼ੀ ਵੱਖਰੇ ਹਨ। ਬਹੁਤ ਸਾਰੇ ਵੱਖ-ਵੱਖ ਅਤੇ ਬਹੁਤ ਖਾਸ ਸਰੋਤਾਂ ਤੋਂ ਮੁੱਖ DWH ਵਿੱਚ ਡੇਟਾ ਲੋਡ ਕਰਨ ਦੇ ਕੰਮ ਹਨ, ਮੁੱਖ DWH ਦੇ ਅੰਦਰ ਸਟੋਰਫਰੰਟ ਦੀ ਗਣਨਾ ਕਰਨ ਦੇ ਕੰਮ ਹਨ, ਇੱਕ ਤੇਜ਼ DWH ਵਿੱਚ ਡੇਟਾ ਪ੍ਰਕਾਸ਼ਿਤ ਕਰਨ ਦੇ ਕੰਮ ਹਨ, ਬਹੁਤ ਸਾਰੇ, ਬਹੁਤ ਸਾਰੇ ਵੱਖ-ਵੱਖ ਕਾਰਜ ਹਨ - ਅਤੇ ਏਅਰਫਲੋ ਦਿਨ-ਪ੍ਰਤੀ-ਦਿਨ ਉਨ੍ਹਾਂ ਨੂੰ ਚਬਾਉਂਦਾ ਹੈ। ਸੰਖਿਆ ਵਿੱਚ ਬੋਲਣਾ, ਇਹ ਹੈ 2,3 ਹਜ਼ਾਰ DWH (Hadoop) ਦੇ ਅੰਦਰ ਵੱਖ-ਵੱਖ ਜਟਿਲਤਾ ਦੇ ELT ਕਾਰਜ, ਲਗਭਗ। 2,5 ਸੌ ਡਾਟਾਬੇਸ ਸਰੋਤ, ਇਹ ਇੱਕ ਟੀਮ ਹੈ 4 ETL ਡਿਵੈਲਪਰ, ਜੋ ਕਿ DWH ਵਿੱਚ ETL ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਅਤੇ DWH ਦੇ ਅੰਦਰ ELT ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਵਿੱਚ ਵੰਡਿਆ ਗਿਆ ਹੈ ਅਤੇ ਬੇਸ਼ਕ ਹੋਰ ਇੱਕ ਪ੍ਰਸ਼ਾਸਕ, ਜੋ ਸੇਵਾ ਦੇ ਬੁਨਿਆਦੀ ਢਾਂਚੇ ਨਾਲ ਨਜਿੱਠਦਾ ਹੈ।

ਭਵਿੱਖ ਲਈ ਯੋਜਨਾਵਾਂ

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

ਏਪੀਲਾਗ

ਇਹ, ਬੇਸ਼ੱਕ, ਉਹ ਸਭ ਕੁਝ ਨਹੀਂ ਹੈ ਜੋ ਮੈਂ ਏਅਰਫਲੋ ਬਾਰੇ ਦੱਸਣਾ ਚਾਹੁੰਦਾ ਹਾਂ, ਪਰ ਮੈਂ ਮੁੱਖ ਨੁਕਤਿਆਂ ਨੂੰ ਉਜਾਗਰ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕੀਤੀ. ਭੁੱਖ ਖਾਣ ਨਾਲ ਲਗਦੀ ਹੈ, ਇਸਨੂੰ ਅਜ਼ਮਾਓ ਅਤੇ ਤੁਹਾਨੂੰ ਇਹ ਪਸੰਦ ਆਵੇਗਾ :)

ਸਰੋਤ: www.habr.com