ಗಾಳಿಯ ಹರಿವು ಬ್ಯಾಚ್ ಡೇಟಾ ಸಂಸ್ಕರಣಾ ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ಅನುಕೂಲಕರವಾಗಿ ಮತ್ತು ತ್ವರಿತವಾಗಿ ಅಭಿವೃದ್ಧಿಪಡಿಸುವ ಮತ್ತು ನಿರ್ವಹಿಸುವ ಸಾಧನವಾಗಿದೆ

ಹಲೋ, ಹಬ್ರ್! ಈ ಲೇಖನದಲ್ಲಿ ನಾನು ಬ್ಯಾಚ್ ಡೇಟಾ ಸಂಸ್ಕರಣಾ ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ಅಭಿವೃದ್ಧಿಪಡಿಸಲು ಒಂದು ಉತ್ತಮ ಸಾಧನದ ಬಗ್ಗೆ ಮಾತನಾಡಲು ಬಯಸುತ್ತೇನೆ, ಉದಾಹರಣೆಗೆ, ಕಾರ್ಪೊರೇಟ್ DWH ಅಥವಾ ನಿಮ್ಮ DataLake ಮೂಲಸೌಕರ್ಯದಲ್ಲಿ. ನಾವು ಅಪಾಚೆ ಗಾಳಿಯ ಹರಿವಿನ ಬಗ್ಗೆ ಮಾತನಾಡುತ್ತೇವೆ (ಇನ್ನು ಮುಂದೆ ಗಾಳಿಯ ಹರಿವು ಎಂದು ಉಲ್ಲೇಖಿಸಲಾಗುತ್ತದೆ). ಇದು ಹ್ಯಾಬ್ರೆಯಲ್ಲಿನ ಗಮನದಿಂದ ಅನ್ಯಾಯವಾಗಿ ವಂಚಿತವಾಗಿದೆ ಮತ್ತು ಮುಖ್ಯ ಭಾಗದಲ್ಲಿ ನಿಮ್ಮ ETL/ELT ಪ್ರಕ್ರಿಯೆಗಳಿಗೆ ಶೆಡ್ಯೂಲರ್ ಅನ್ನು ಆಯ್ಕೆಮಾಡುವಾಗ ಕನಿಷ್ಠ ಗಾಳಿಯ ಹರಿವು ನೋಡುವುದು ಯೋಗ್ಯವಾಗಿದೆ ಎಂದು ನಾನು ನಿಮಗೆ ಮನವರಿಕೆ ಮಾಡಲು ಪ್ರಯತ್ನಿಸುತ್ತೇನೆ.

ಹಿಂದೆ, ನಾನು ಟಿಂಕಾಫ್ ಬ್ಯಾಂಕ್‌ನಲ್ಲಿ ಕೆಲಸ ಮಾಡುವಾಗ DWH ವಿಷಯದ ಕುರಿತು ಲೇಖನಗಳ ಸರಣಿಯನ್ನು ಬರೆದಿದ್ದೇನೆ. ಈಗ ನಾನು Mail.Ru ಗ್ರೂಪ್ ತಂಡದ ಭಾಗವಾಗಿದ್ದೇನೆ ಮತ್ತು ಗೇಮಿಂಗ್ ಪ್ರದೇಶದಲ್ಲಿ ಡೇಟಾ ವಿಶ್ಲೇಷಣೆಗಾಗಿ ವೇದಿಕೆಯನ್ನು ಅಭಿವೃದ್ಧಿಪಡಿಸುತ್ತಿದ್ದೇನೆ. ವಾಸ್ತವವಾಗಿ, ಸುದ್ದಿ ಮತ್ತು ಆಸಕ್ತಿದಾಯಕ ಪರಿಹಾರಗಳು ಕಾಣಿಸಿಕೊಂಡಂತೆ, ಡೇಟಾ ವಿಶ್ಲೇಷಣೆಗಾಗಿ ನಮ್ಮ ಪ್ಲಾಟ್‌ಫಾರ್ಮ್ ಕುರಿತು ನನ್ನ ತಂಡ ಮತ್ತು ನಾನು ಇಲ್ಲಿ ಮಾತನಾಡುತ್ತೇವೆ.

ಮುನ್ನುಡಿ

ಆದ್ದರಿಂದ, ಪ್ರಾರಂಭಿಸೋಣ. ಗಾಳಿಯ ಹರಿವು ಎಂದರೇನು? ಇದು ಗ್ರಂಥಾಲಯವಾಗಿದೆ (ಅಥವಾ ಗ್ರಂಥಾಲಯಗಳ ಸೆಟ್) ಕೆಲಸದ ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ಅಭಿವೃದ್ಧಿಪಡಿಸಲು, ಯೋಜಿಸಲು ಮತ್ತು ಮೇಲ್ವಿಚಾರಣೆ ಮಾಡಲು. ಗಾಳಿಯ ಹರಿವಿನ ಮುಖ್ಯ ಲಕ್ಷಣ: ಪೈಥಾನ್ ಕೋಡ್ ಅನ್ನು ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ವಿವರಿಸಲು (ಅಭಿವೃದ್ಧಿಪಡಿಸಲು) ಬಳಸಲಾಗುತ್ತದೆ. ನಿಮ್ಮ ಯೋಜನೆ ಮತ್ತು ಅಭಿವೃದ್ಧಿಯನ್ನು ಸಂಘಟಿಸಲು ಇದು ಬಹಳಷ್ಟು ಪ್ರಯೋಜನಗಳನ್ನು ಹೊಂದಿದೆ: ಮೂಲಭೂತವಾಗಿ, ನಿಮ್ಮ (ಉದಾಹರಣೆಗೆ) ಇಟಿಎಲ್ ಯೋಜನೆಯು ಕೇವಲ ಪೈಥಾನ್ ಯೋಜನೆಯಾಗಿದೆ ಮತ್ತು ಮೂಲಸೌಕರ್ಯ, ತಂಡದ ಗಾತ್ರ ಮತ್ತು ನಿರ್ದಿಷ್ಟತೆಗಳನ್ನು ಗಣನೆಗೆ ತೆಗೆದುಕೊಂಡು ನೀವು ಬಯಸಿದಂತೆ ನೀವು ಅದನ್ನು ಸಂಘಟಿಸಬಹುದು. ಇತರ ಅವಶ್ಯಕತೆಗಳು. ವಾದ್ಯವಾಗಿ ಎಲ್ಲವೂ ಸರಳವಾಗಿದೆ. ಉದಾಹರಣೆಗೆ PyCharm + Git ಬಳಸಿ. ಇದು ಅದ್ಭುತವಾಗಿದೆ ಮತ್ತು ತುಂಬಾ ಅನುಕೂಲಕರವಾಗಿದೆ!

ಈಗ ಗಾಳಿಯ ಹರಿವಿನ ಮುಖ್ಯ ಘಟಕಗಳನ್ನು ನೋಡೋಣ. ಅವುಗಳ ಸಾರ ಮತ್ತು ಉದ್ದೇಶವನ್ನು ಅರ್ಥಮಾಡಿಕೊಳ್ಳುವ ಮೂಲಕ, ನಿಮ್ಮ ಪ್ರಕ್ರಿಯೆಯ ವಾಸ್ತುಶಿಲ್ಪವನ್ನು ನೀವು ಅತ್ಯುತ್ತಮವಾಗಿ ಸಂಘಟಿಸಬಹುದು. ಬಹುಶಃ ಮುಖ್ಯ ಘಟಕವು ಡೈರೆಕ್ಟೆಡ್ ಅಸಿಕ್ಲಿಕ್ ಗ್ರಾಫ್ ಆಗಿರಬಹುದು (ಇನ್ನು ಮುಂದೆ DAG ಎಂದು ಉಲ್ಲೇಖಿಸಲಾಗುತ್ತದೆ).

DAG

DAG ಎನ್ನುವುದು ನಿಮ್ಮ ಕಾರ್ಯಗಳ ಕೆಲವು ಅರ್ಥಪೂರ್ಣ ಸಂಯೋಜನೆಯಾಗಿದ್ದು, ನಿರ್ದಿಷ್ಟ ವೇಳಾಪಟ್ಟಿಯ ಪ್ರಕಾರ ಕಟ್ಟುನಿಟ್ಟಾಗಿ ವ್ಯಾಖ್ಯಾನಿಸಲಾದ ಅನುಕ್ರಮದಲ್ಲಿ ನೀವು ಪೂರ್ಣಗೊಳಿಸಲು ಬಯಸುತ್ತೀರಿ. DAG ಗಳು ಮತ್ತು ಇತರ ಘಟಕಗಳೊಂದಿಗೆ ಕೆಲಸ ಮಾಡಲು ಏರ್‌ಫ್ಲೋ ಅನುಕೂಲಕರ ವೆಬ್ ಇಂಟರ್ಫೇಸ್ ಅನ್ನು ಒದಗಿಸುತ್ತದೆ:

DAG ಈ ರೀತಿ ಕಾಣಿಸಬಹುದು:

ಡೆವಲಪರ್, DAG ಅನ್ನು ವಿನ್ಯಾಸಗೊಳಿಸುವಾಗ, DAG ಯೊಳಗಿನ ಕಾರ್ಯಗಳನ್ನು ನಿರ್ಮಿಸುವ ನಿರ್ವಾಹಕರ ಗುಂಪನ್ನು ಇಡುತ್ತಾರೆ. ಇಲ್ಲಿ ನಾವು ಮತ್ತೊಂದು ಪ್ರಮುಖ ಘಟಕಕ್ಕೆ ಬರುತ್ತೇವೆ: ಏರ್‌ಫ್ಲೋ ಆಪರೇಟರ್.

ಆಪರೇಟರ್ಗಳು

ಆಪರೇಟರ್ ಎನ್ನುವುದು ಉದ್ಯೋಗದ ನಿದರ್ಶನಗಳನ್ನು ರಚಿಸುವ ಆಧಾರದ ಮೇಲೆ ಒಂದು ಘಟಕವಾಗಿದೆ, ಇದು ಕೆಲಸದ ನಿದರ್ಶನವನ್ನು ಕಾರ್ಯಗತಗೊಳಿಸುವಾಗ ಏನಾಗುತ್ತದೆ ಎಂಬುದನ್ನು ವಿವರಿಸುತ್ತದೆ. GitHub ನಿಂದ ಗಾಳಿಯ ಹರಿವು ಬಿಡುಗಡೆಯಾಗುತ್ತದೆ ಈಗಾಗಲೇ ಬಳಸಲು ಸಿದ್ಧವಾಗಿರುವ ನಿರ್ವಾಹಕರ ಗುಂಪನ್ನು ಒಳಗೊಂಡಿದೆ. ಉದಾಹರಣೆಗಳು:

• BashOperator - ಬ್ಯಾಷ್ ಆಜ್ಞೆಯನ್ನು ಕಾರ್ಯಗತಗೊಳಿಸಲು ಆಪರೇಟರ್.
• ಪೈಥಾನ್ ಆಪರೇಟರ್ - ಪೈಥಾನ್ ಕೋಡ್ ಕರೆ ಮಾಡಲು ಆಪರೇಟರ್.
• ಇಮೇಲ್ ಆಪರೇಟರ್ - ಇಮೇಲ್ ಕಳುಹಿಸಲು ಆಪರೇಟರ್.
• HTTPOoperator - http ವಿನಂತಿಗಳೊಂದಿಗೆ ಕೆಲಸ ಮಾಡಲು ಆಪರೇಟರ್.
• SqlOperator - SQL ಕೋಡ್ ಅನ್ನು ಕಾರ್ಯಗತಗೊಳಿಸಲು ಆಪರೇಟರ್.
• ಸಂವೇದಕವು ಈವೆಂಟ್‌ಗಾಗಿ ಕಾಯುವ ಆಪರೇಟರ್ ಆಗಿದೆ (ಅಗತ್ಯವಿರುವ ಸಮಯದ ಆಗಮನ, ಅಗತ್ಯವಿರುವ ಫೈಲ್‌ನ ನೋಟ, ಡೇಟಾಬೇಸ್‌ನಲ್ಲಿ ಒಂದು ಸಾಲು, API ಯಿಂದ ಪ್ರತಿಕ್ರಿಯೆ, ಇತ್ಯಾದಿ. ಇತ್ಯಾದಿ).

ಹೆಚ್ಚು ನಿರ್ದಿಷ್ಟ ನಿರ್ವಾಹಕರು ಇದ್ದಾರೆ: ಡಾಕರ್ ಆಪರೇಟರ್, ಹೈವ್ ಆಪರೇಟರ್, ಎಸ್3ಫೈಲ್ ಟ್ರಾನ್ಸ್‌ಫರ್ ಆಪರೇಟರ್, ಪ್ರೆಸ್ಟೋಟೊಮೈಸ್ಕ್ಲ್ ಆಪರೇಟರ್, ಸ್ಲಾಕ್ ಆಪರೇಟರ್.

ನಿಮ್ಮ ಸ್ವಂತ ಗುಣಲಕ್ಷಣಗಳ ಆಧಾರದ ಮೇಲೆ ನೀವು ನಿರ್ವಾಹಕರನ್ನು ಅಭಿವೃದ್ಧಿಪಡಿಸಬಹುದು ಮತ್ತು ಅವುಗಳನ್ನು ನಿಮ್ಮ ಯೋಜನೆಯಲ್ಲಿ ಬಳಸಬಹುದು. ಉದಾಹರಣೆಗೆ, ನಾವು MongoDBToHiveViaHdfsTransfer ಅನ್ನು ರಚಿಸಿದ್ದೇವೆ, MongoDB ಯಿಂದ ಹೈವ್‌ಗೆ ಡಾಕ್ಯುಮೆಂಟ್‌ಗಳನ್ನು ರಫ್ತು ಮಾಡುವ ಆಪರೇಟರ್ ಮತ್ತು ಕೆಲಸ ಮಾಡಲು ಹಲವಾರು ಆಪರೇಟರ್‌ಗಳು ಕ್ಲಿಕ್‌ಹೌಸ್: 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. ಹಿಂದಿನ ಕಾರ್ಯಗಳನ್ನು DAG ನಲ್ಲಿ ಪೂರ್ಣಗೊಳಿಸಲಾಗಿದೆ; ಹೊಸದನ್ನು ಸರದಿಯಲ್ಲಿ ಇರಿಸಬಹುದು.
2. ಕಾರ್ಯಗಳ ಆದ್ಯತೆಯನ್ನು ಅವಲಂಬಿಸಿ ಕ್ಯೂ ಅನ್ನು ವಿಂಗಡಿಸಲಾಗಿದೆ (ಆದ್ಯತೆಗಳನ್ನು ಸಹ ನಿಯಂತ್ರಿಸಬಹುದು), ಮತ್ತು ಪೂಲ್ನಲ್ಲಿ ಉಚಿತ ಸ್ಲಾಟ್ ಇದ್ದರೆ, ಕಾರ್ಯವನ್ನು ಕಾರ್ಯಾಚರಣೆಗೆ ತೆಗೆದುಕೊಳ್ಳಬಹುದು.
3. ಉಚಿತ ಕೆಲಸಗಾರ ಸೆಲರಿ ಇದ್ದರೆ, ಕೆಲಸವನ್ನು ಅದಕ್ಕೆ ಕಳುಹಿಸಲಾಗುತ್ತದೆ; ಸಮಸ್ಯೆಯಲ್ಲಿ ನೀವು ಪ್ರೋಗ್ರಾಮ್ ಮಾಡಿದ ಕೆಲಸವು ಒಂದು ಅಥವಾ ಇನ್ನೊಂದು ಆಪರೇಟರ್ ಅನ್ನು ಬಳಸಿಕೊಂಡು ಪ್ರಾರಂಭವಾಗುತ್ತದೆ.

ಸಾಕಷ್ಟು ಸರಳ.

ಶೆಡ್ಯೂಲರ್ ಎಲ್ಲಾ DAG ಗಳ ಸೆಟ್ ಮತ್ತು DAG ಗಳೊಳಗಿನ ಎಲ್ಲಾ ಕಾರ್ಯಗಳ ಮೇಲೆ ರನ್ ಆಗುತ್ತದೆ.

ಶೆಡ್ಯೂಲರ್ 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 ಕಾರ್ಯಗತಗೊಳಿಸುವ ದಿನಾಂಕದ ಆಯಾಮವನ್ನು ಹೊಂದಿದೆ, ಅಂದರೆ, DAG ಯ ಕೆಲಸದ ವೇಳಾಪಟ್ಟಿಯನ್ನು ಅವಲಂಬಿಸಿ, ಪ್ರತಿ ಎಕ್ಸಿಕ್ಯೂಶನ್ ದಿನಾಂಕಕ್ಕೆ ಕಾರ್ಯ ನಿದರ್ಶನಗಳನ್ನು ರಚಿಸಲಾಗುತ್ತದೆ. ಮತ್ತು ಪ್ರತಿ ಎಕ್ಸಿಕ್ಯೂಶನ್ ದಿನಾಂಕಕ್ಕೆ, ಕಾರ್ಯಗಳನ್ನು ಮರು-ಕಾರ್ಯಗತಗೊಳಿಸಬಹುದು - ಅಥವಾ, ಉದಾಹರಣೆಗೆ, DAG ಹಲವಾರು ಎಕ್ಸಿಕ್ಯೂಶನ್ ದಿನಾಂಕಗಳಲ್ಲಿ ಏಕಕಾಲದಲ್ಲಿ ಕೆಲಸ ಮಾಡಬಹುದು. ಇದನ್ನು ಇಲ್ಲಿ ಸ್ಪಷ್ಟವಾಗಿ ತೋರಿಸಲಾಗಿದೆ:

ದುರದೃಷ್ಟವಶಾತ್ (ಅಥವಾ ಬಹುಶಃ ಅದೃಷ್ಟವಶಾತ್: ಇದು ಪರಿಸ್ಥಿತಿಯನ್ನು ಅವಲಂಬಿಸಿರುತ್ತದೆ), DAG ಯಲ್ಲಿನ ಕಾರ್ಯದ ಅನುಷ್ಠಾನವನ್ನು ಸರಿಪಡಿಸಿದರೆ, ಹಿಂದಿನ ಮರಣದಂಡನೆ ದಿನಾಂಕದಲ್ಲಿ ಮರಣದಂಡನೆಯು ಹೊಂದಾಣಿಕೆಗಳನ್ನು ಗಣನೆಗೆ ತೆಗೆದುಕೊಂಡು ಮುಂದುವರಿಯುತ್ತದೆ. ಹೊಸ ಅಲ್ಗಾರಿದಮ್ ಬಳಸಿ ನೀವು ಹಿಂದಿನ ಅವಧಿಗಳಲ್ಲಿ ಡೇಟಾವನ್ನು ಮರು ಲೆಕ್ಕಾಚಾರ ಮಾಡಬೇಕಾದರೆ ಇದು ಒಳ್ಳೆಯದು, ಆದರೆ ಫಲಿತಾಂಶದ ಪುನರುತ್ಪಾದನೆಯು ಕಳೆದುಹೋದ ಕಾರಣ ಅದು ಕೆಟ್ಟದಾಗಿದೆ (ಸಹಜವಾಗಿ, Git ನಿಂದ ಮೂಲ ಕೋಡ್‌ನ ಅಗತ್ಯವಿರುವ ಆವೃತ್ತಿಯನ್ನು ಹಿಂತಿರುಗಿಸಲು ಮತ್ತು ಏನನ್ನು ಲೆಕ್ಕಹಾಕಲು ಯಾರೂ ನಿಮಗೆ ತೊಂದರೆ ಕೊಡುವುದಿಲ್ಲ. ನಿಮಗೆ ಒಂದು ಬಾರಿ ಬೇಕು, ನಿಮಗೆ ಬೇಕಾದ ರೀತಿಯಲ್ಲಿ).

ಕಾರ್ಯಗಳನ್ನು ರಚಿಸುವುದು

DAG ಯ ಅನುಷ್ಠಾನವು ಪೈಥಾನ್‌ನಲ್ಲಿ ಕೋಡ್ ಆಗಿದೆ, ಆದ್ದರಿಂದ ಕೆಲಸ ಮಾಡುವಾಗ ಕೋಡ್‌ನ ಪ್ರಮಾಣವನ್ನು ಕಡಿಮೆ ಮಾಡಲು ನಾವು ತುಂಬಾ ಅನುಕೂಲಕರ ಮಾರ್ಗವನ್ನು ಹೊಂದಿದ್ದೇವೆ, ಉದಾಹರಣೆಗೆ, ಚೂರುಚೂರು ಮೂಲಗಳೊಂದಿಗೆ. ನೀವು ಮೂಲವಾಗಿ ಮೂರು 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 ಈ ರೀತಿ ಕಾಣುತ್ತದೆ:

ಈ ಸಂದರ್ಭದಲ್ಲಿ, ಸೆಟ್ಟಿಂಗ್‌ಗಳನ್ನು ಸರಳವಾಗಿ ಹೊಂದಿಸುವ ಮೂಲಕ ಮತ್ತು DAG ಅನ್ನು ನವೀಕರಿಸುವ ಮೂಲಕ ನೀವು ಚೂರುಗಳನ್ನು ಸೇರಿಸಬಹುದು ಅಥವಾ ತೆಗೆದುಹಾಕಬಹುದು. ಆರಾಮದಾಯಕ!

ನೀವು ಹೆಚ್ಚು ಸಂಕೀರ್ಣವಾದ ಕೋಡ್ ಉತ್ಪಾದನೆಯನ್ನು ಸಹ ಬಳಸಬಹುದು, ಉದಾಹರಣೆಗೆ, ಡೇಟಾಬೇಸ್ ರೂಪದಲ್ಲಿ ಮೂಲಗಳೊಂದಿಗೆ ಕೆಲಸ ಮಾಡಿ ಅಥವಾ ಟೇಬಲ್ ರಚನೆಯನ್ನು ವಿವರಿಸಿ, ಟೇಬಲ್‌ನೊಂದಿಗೆ ಕೆಲಸ ಮಾಡುವ ಅಲ್ಗಾರಿದಮ್, ಮತ್ತು, DWH ಮೂಲಸೌಕರ್ಯದ ವೈಶಿಷ್ಟ್ಯಗಳನ್ನು ಗಣನೆಗೆ ತೆಗೆದುಕೊಂಡು, ಪ್ರಕ್ರಿಯೆಯನ್ನು ರಚಿಸಿ ನಿಮ್ಮ ಸಂಗ್ರಹಣೆಗೆ N ಕೋಷ್ಟಕಗಳನ್ನು ಲೋಡ್ ಮಾಡಲು. ಅಥವಾ, ಉದಾಹರಣೆಗೆ, ಪಟ್ಟಿಯ ರೂಪದಲ್ಲಿ ಪ್ಯಾರಾಮೀಟರ್‌ನೊಂದಿಗೆ ಕಾರ್ಯನಿರ್ವಹಿಸುವುದನ್ನು ಬೆಂಬಲಿಸದ API ನೊಂದಿಗೆ ಕೆಲಸ ಮಾಡುವುದು, ನೀವು ಈ ಪಟ್ಟಿಯಿಂದ DAG ನಲ್ಲಿ N ಕಾರ್ಯಗಳನ್ನು ರಚಿಸಬಹುದು, API ನಲ್ಲಿ ವಿನಂತಿಗಳ ಸಮಾನಾಂತರತೆಯನ್ನು ಪೂಲ್‌ಗೆ ಮಿತಿಗೊಳಿಸಬಹುದು ಮತ್ತು ಸ್ಕ್ರ್ಯಾಪ್ ಮಾಡಬಹುದು API ನಿಂದ ಅಗತ್ಯ ಡೇಟಾ. ಹೊಂದಿಕೊಳ್ಳುವ!

ಭಂಡಾರ

ಏರ್‌ಫ್ಲೋ ತನ್ನದೇ ಆದ ಬ್ಯಾಕೆಂಡ್ ರೆಪೊಸಿಟರಿಯನ್ನು ಹೊಂದಿದೆ, ಡೇಟಾಬೇಸ್ (MySQL ಅಥವಾ ಪೋಸ್ಟ್‌ಗ್ರೆಸ್ ಆಗಿರಬಹುದು, ನಾವು ಪೋಸ್ಟ್‌ಗ್ರೆಸ್ ಅನ್ನು ಹೊಂದಿದ್ದೇವೆ), ಇದು ಕಾರ್ಯಗಳ ಸ್ಥಿತಿಗಳು, DAG ಗಳು, ಸಂಪರ್ಕ ಸೆಟ್ಟಿಂಗ್‌ಗಳು, ಜಾಗತಿಕ ವೇರಿಯಬಲ್‌ಗಳು ಇತ್ಯಾದಿಗಳನ್ನು ಸಂಗ್ರಹಿಸುತ್ತದೆ. ಇಲ್ಲಿ ನಾನು ಹೇಳಲು ಬಯಸುತ್ತೇನೆ ಏರ್‌ಫ್ಲೋದಲ್ಲಿನ ರೆಪೊಸಿಟರಿಯು ತುಂಬಾ ಸರಳವಾಗಿದೆ (ಸುಮಾರು 20 ಕೋಷ್ಟಕಗಳು) ಮತ್ತು ನೀವು ಅದರ ಮೇಲೆ ನಿಮ್ಮದೇ ಆದ ಯಾವುದೇ ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ನಿರ್ಮಿಸಲು ಬಯಸಿದರೆ ಅನುಕೂಲಕರವಾಗಿದೆ. ಇನ್ಫರ್ಮ್ಯಾಟಿಕಾ ರೆಪೊಸಿಟರಿಯಲ್ಲಿನ 100500 ಕೋಷ್ಟಕಗಳನ್ನು ನಾನು ನೆನಪಿಸಿಕೊಳ್ಳುತ್ತೇನೆ, ಪ್ರಶ್ನೆಯನ್ನು ಹೇಗೆ ನಿರ್ಮಿಸುವುದು ಎಂಬುದನ್ನು ಅರ್ಥಮಾಡಿಕೊಳ್ಳುವ ಮೊದಲು ದೀರ್ಘಕಾಲದವರೆಗೆ ಅಧ್ಯಯನ ಮಾಡಬೇಕಾಗಿತ್ತು.

ಮಾನಿಟರಿಂಗ್

ರೆಪೊಸಿಟರಿಯ ಸರಳತೆಯನ್ನು ನೀಡಿದರೆ, ನಿಮಗೆ ಅನುಕೂಲಕರವಾದ ಕಾರ್ಯ ಮೇಲ್ವಿಚಾರಣೆ ಪ್ರಕ್ರಿಯೆಯನ್ನು ನೀವು ನಿರ್ಮಿಸಬಹುದು. ನಾವು ಜೆಪ್ಪೆಲಿನ್‌ನಲ್ಲಿ ನೋಟ್‌ಪ್ಯಾಡ್ ಅನ್ನು ಬಳಸುತ್ತೇವೆ, ಅಲ್ಲಿ ನಾವು ಕಾರ್ಯಗಳ ಸ್ಥಿತಿಯನ್ನು ನೋಡುತ್ತೇವೆ:

ಇದು ಏರ್‌ಫ್ಲೋನ ವೆಬ್ ಇಂಟರ್‌ಫೇಸ್ ಆಗಿರಬಹುದು:

ಏರ್‌ಫ್ಲೋ ಕೋಡ್ ಮುಕ್ತ ಮೂಲವಾಗಿದೆ, ಆದ್ದರಿಂದ ನಾವು ಟೆಲಿಗ್ರಾಮ್‌ಗೆ ಎಚ್ಚರಿಕೆಯನ್ನು ಸೇರಿಸಿದ್ದೇವೆ. ಕಾರ್ಯದ ಪ್ರತಿಯೊಂದು ಚಾಲನೆಯಲ್ಲಿರುವ ನಿದರ್ಶನವು ದೋಷ ಸಂಭವಿಸಿದಲ್ಲಿ, ಸಂಪೂರ್ಣ ಅಭಿವೃದ್ಧಿ ಮತ್ತು ಬೆಂಬಲ ತಂಡವನ್ನು ಒಳಗೊಂಡಿರುವ ಟೆಲಿಗ್ರಾಮ್‌ನಲ್ಲಿ ಗುಂಪನ್ನು ಸ್ಪ್ಯಾಮ್ ಮಾಡುತ್ತದೆ.

ನಾವು ಟೆಲಿಗ್ರಾಮ್ ಮೂಲಕ ಪ್ರಾಂಪ್ಟ್ ಪ್ರತಿಕ್ರಿಯೆಯನ್ನು ಸ್ವೀಕರಿಸುತ್ತೇವೆ (ಅಗತ್ಯವಿದ್ದರೆ), ಮತ್ತು ಜೆಪ್ಪೆಲಿನ್ ಮೂಲಕ ನಾವು ಏರ್‌ಫ್ಲೋನಲ್ಲಿನ ಕಾರ್ಯಗಳ ಒಟ್ಟಾರೆ ಚಿತ್ರವನ್ನು ಸ್ವೀಕರಿಸುತ್ತೇವೆ.

ಒಟ್ಟು

ಗಾಳಿಯ ಹರಿವು ಪ್ರಾಥಮಿಕವಾಗಿ ತೆರೆದ ಮೂಲವಾಗಿದೆ ಮತ್ತು ಅದರಿಂದ ನೀವು ಪವಾಡಗಳನ್ನು ನಿರೀಕ್ಷಿಸಬಾರದು. ಕೆಲಸ ಮಾಡುವ ಪರಿಹಾರವನ್ನು ನಿರ್ಮಿಸಲು ಸಮಯ ಮತ್ತು ಶ್ರಮವನ್ನು ಹಾಕಲು ಸಿದ್ಧರಾಗಿರಿ. ಗುರಿಯನ್ನು ಸಾಧಿಸಬಹುದು, ನನ್ನನ್ನು ನಂಬಿರಿ, ಅದು ಯೋಗ್ಯವಾಗಿದೆ. ಅಭಿವೃದ್ಧಿಯ ವೇಗ, ನಮ್ಯತೆ, ಹೊಸ ಪ್ರಕ್ರಿಯೆಗಳನ್ನು ಸೇರಿಸುವ ಸುಲಭ - ನೀವು ಅದನ್ನು ಇಷ್ಟಪಡುತ್ತೀರಿ. ಸಹಜವಾಗಿ, ನೀವು ಯೋಜನೆಯ ಸಂಘಟನೆಗೆ ಹೆಚ್ಚಿನ ಗಮನವನ್ನು ನೀಡಬೇಕಾಗಿದೆ, ಗಾಳಿಯ ಹರಿವಿನ ಸ್ಥಿರತೆ ಸ್ವತಃ: ಪವಾಡಗಳು ಸಂಭವಿಸುವುದಿಲ್ಲ.

ಈಗ ನಾವು ಪ್ರತಿದಿನ ಗಾಳಿಯ ಹರಿವನ್ನು ಮಾಡುತ್ತಿದ್ದೇವೆ ಸುಮಾರು 6,5 ಸಾವಿರ ಕಾರ್ಯಗಳು. ಅವರು ಪಾತ್ರದಲ್ಲಿ ಸಾಕಷ್ಟು ಭಿನ್ನರಾಗಿದ್ದಾರೆ. ಹಲವಾರು ವಿಭಿನ್ನ ಮತ್ತು ನಿರ್ದಿಷ್ಟ ಮೂಲಗಳಿಂದ ಮುಖ್ಯ DWH ಗೆ ಡೇಟಾವನ್ನು ಲೋಡ್ ಮಾಡುವ ಕಾರ್ಯಗಳಿವೆ, ಮುಖ್ಯ DWH ಒಳಗೆ ಅಂಗಡಿ ಮುಂಭಾಗಗಳನ್ನು ಲೆಕ್ಕಾಚಾರ ಮಾಡುವ ಕಾರ್ಯಗಳಿವೆ, ವೇಗದ DWH ಗೆ ಡೇಟಾವನ್ನು ಪ್ರಕಟಿಸುವ ಕಾರ್ಯಗಳಿವೆ, ಹಲವು ವಿಭಿನ್ನ ಕಾರ್ಯಗಳಿವೆ - ಮತ್ತು ಗಾಳಿಯ ಹರಿವು ಅವುಗಳನ್ನು ದಿನದಿಂದ ದಿನಕ್ಕೆ ಅಗಿಯುತ್ತಾನೆ. ಸಂಖ್ಯೆಯಲ್ಲಿ ಹೇಳುವುದಾದರೆ, ಇದು Xnumx ಸಾವಿರ DWH (ಹಡೂಪ್) ಒಳಗೆ ವಿಭಿನ್ನ ಸಂಕೀರ್ಣತೆಯ ELT ಕಾರ್ಯಗಳು, ಅಂದಾಜು. 2,5 ನೂರು ಡೇಟಾಬೇಸ್ ಮೂಲಗಳು, ಇದು ತಂಡವಾಗಿದೆ 4 ಇಟಿಎಲ್ ಡೆವಲಪರ್‌ಗಳು, ಇವುಗಳನ್ನು DWH ನಲ್ಲಿ ETL ಡೇಟಾ ಸಂಸ್ಕರಣೆ ಮತ್ತು DWH ಒಳಗೆ ELT ಡೇಟಾ ಪ್ರಕ್ರಿಯೆಗೆ ವಿಂಗಡಿಸಲಾಗಿದೆ ಮತ್ತು ಸಹಜವಾಗಿ ಹೆಚ್ಚು ಒಬ್ಬ ನಿರ್ವಾಹಕ, ಯಾರು ಸೇವೆಯ ಮೂಲಸೌಕರ್ಯದೊಂದಿಗೆ ವ್ಯವಹರಿಸುತ್ತಾರೆ.

ಭವಿಷ್ಯದ ಯೋಜನೆಗಳು

ಪ್ರಕ್ರಿಯೆಗಳ ಸಂಖ್ಯೆಯು ಅನಿವಾರ್ಯವಾಗಿ ಬೆಳೆಯುತ್ತಿದೆ ಮತ್ತು ಗಾಳಿಯ ಹರಿವಿನ ಮೂಲಸೌಕರ್ಯದ ವಿಷಯದಲ್ಲಿ ನಾವು ಮಾಡುವ ಮುಖ್ಯ ವಿಷಯವೆಂದರೆ ಸ್ಕೇಲಿಂಗ್. ನಾವು ಏರ್‌ಫ್ಲೋ ಕ್ಲಸ್ಟರ್ ಅನ್ನು ನಿರ್ಮಿಸಲು ಬಯಸುತ್ತೇವೆ, ಸೆಲರಿ ಕೆಲಸಗಾರರಿಗೆ ಒಂದು ಜೋಡಿ ಕಾಲುಗಳನ್ನು ನಿಯೋಜಿಸುತ್ತೇವೆ ಮತ್ತು ಕೆಲಸದ ವೇಳಾಪಟ್ಟಿ ಪ್ರಕ್ರಿಯೆಗಳು ಮತ್ತು ರೆಪೊಸಿಟರಿಯೊಂದಿಗೆ ಸ್ವಯಂ-ನಕಲು ಮಾಡುವ ತಲೆಯನ್ನು ಮಾಡಲು ಬಯಸುತ್ತೇವೆ.

ಸಂಚಿಕೆ

ಇದು ಸಹಜವಾಗಿ, ನಾನು ಗಾಳಿಯ ಹರಿವಿನ ಬಗ್ಗೆ ಹೇಳಲು ಬಯಸುವ ಎಲ್ಲವೂ ಅಲ್ಲ, ಆದರೆ ನಾನು ಮುಖ್ಯ ಅಂಶಗಳನ್ನು ಹೈಲೈಟ್ ಮಾಡಲು ಪ್ರಯತ್ನಿಸಿದೆ. ತಿನ್ನುವುದರೊಂದಿಗೆ ಹಸಿವು ಬರುತ್ತದೆ, ಇದನ್ನು ಪ್ರಯತ್ನಿಸಿ ಮತ್ತು ನೀವು ಅದನ್ನು ಇಷ್ಟಪಡುತ್ತೀರಿ :)

ಮೂಲ: www.habr.com