{"id":2763,"date":"2024-10-26T06:49:11","date_gmt":"2024-10-25T22:49:11","guid":{"rendered":"http:\/\/viplao.com\/?p=2763"},"modified":"2024-10-26T06:52:22","modified_gmt":"2024-10-25T22:52:22","slug":"%e5%a4%a7%e8%a7%84%e6%a8%a1%e8%bf%90%e8%a1%8c-apache-airflow-%e7%9a%84%e7%bb%8f%e9%aa%8c%e5%92%8c%e6%95%99%e8%ae%ad","status":"publish","type":"post","link":"http:\/\/viplao.com\/index.php\/2024\/10\/26\/%e5%a4%a7%e8%a7%84%e6%a8%a1%e8%bf%90%e8%a1%8c-apache-airflow-%e7%9a%84%e7%bb%8f%e9%aa%8c%e5%92%8c%e6%95%99%e8%ae%ad\/","title":{"rendered":"\u5927\u89c4\u6a21\u8fd0\u884c Apache Airflow \u7684\u7ecf\u9a8c\u548c\u6559\u8bad"},"content":{"rendered":"\n

Apache Airflow \u662f\u4e00\u4e2a\u80fd\u591f\u5f00\u53d1\u3001\u8c03\u5ea6\u548c\u76d1\u63a7\u5de5\u4f5c\u6d41\u7684\u7f16\u6392\u5e73\u53f0\u3002\u5728 Shopify\uff0c\u6211\u4eec\u5df2\u7ecf\u5728\u751f\u4ea7\u4e2d\u8fd0\u884c\u4e86\u4e24\u5e74\u591a\u7684 Airflow\uff0c\u7528\u4e8e\u5404\u79cd\u5de5\u4f5c\u6d41\uff0c\u5305\u62ec\u6570\u636e\u63d0\u53d6\u3001\u673a\u5668\u5b66\u4e60\u6a21\u578b\u8bad\u7ec3\u3001Apache Iceberg \u8868\u7ef4\u62a4\u548c DBT \u9a71\u52a8\u7684\u6570\u636e\u5efa\u6a21\u3002\u5728\u64b0\u5199\u672c\u6587\u65f6\uff0c\u6211\u4eec\u6b63\u901a\u8fc7 Celery \u6267\u884c\u5668\u548c MySQL 8 \u5728 Kubernetes \u4e0a\u6765\u8fd0\u884c Airflow 2.2\u3002<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

Shopify \u5728 Airflow \u4e0a\u7684\u5e94\u7528\u89c4\u6a21\u5728\u8fc7\u53bb\u4e24\u5e74\u4e2d\u6025\u5267\u6269\u5927\u3002\u5728\u6211\u4eec\u6700\u5927\u7684\u5e94\u7528\u573a\u666f\u4e2d\uff0c\u6211\u4eec\u4f7f\u7528\u4e86 10000 \u591a\u4e2a DAG\uff0c\u4ee3\u8868\u4e86\u5927\u91cf\u4e0d\u540c\u7684\u5de5\u4f5c\u8d1f\u8f7d\u3002\u5728\u8fd9\u4e2a\u573a\u666f\u4e2d\uff0c\u5e73\u5747\u6709 400 \u591a\u9879\u4efb\u52a1\u6b63\u5728\u8fdb\u884c\uff0c\u5e76\u4e14\u6bcf\u5929\u7684\u8fd0\u884c\u6b21\u6570\u8d85\u8fc7 14 \u4e07\u6b21\u3002\u7531\u4e8e Shopify \u7684\u5185\u90e8\u91c7\u7528\u7387\u8d8a\u6765\u8d8a\u9ad8\uff0c\u6211\u4eec\u7684 Airflow \u90e8\u7f72\u5c06\u4f1a\u4ea7\u751f\u66f4\u591a\u7684\u8d1f\u8f7d\u3002\u56e0\u4e3a\u8fd9\u6837\u7684\u8fc5\u901f\u589e\u957f\uff0c\u6211\u4eec\u6240\u9762\u4e34\u7684\u56f0\u96be\u5305\u62ec\uff1a\u6587\u4ef6\u5b58\u53d6\u901f\u5ea6\u592a\u6162\u3001\u5bf9 DAG\uff08Directed acyclic graph\uff0c\u6709\u5411\u65e0\u73af\u56fe\uff09\u80fd\u529b\u7684\u63a7\u5236\u4e0d\u8db3\u3001\u6d41\u91cf\u6c34\u5e73\u7684\u4e0d\u89c4\u5219\u3001\u5de5\u4f5c\u8d1f\u8f7d\u4e4b\u95f4\u7684\u8d44\u6e90\u4e89\u7528\u7b49\u7b49\u3002<\/p>\n\n\n\n

\u63a5\u4e0b\u6765\uff0c\u6211\u4eec\u5c06\u4e0e\u5927\u5bb6\u5206\u4eab\u6211\u4eec\u6240\u83b7\u5f97\u7684\u7ecf\u9a8c\u4ee5\u53ca\u6211\u4eec\u4e3a\u5b9e\u73b0\u5927\u89c4\u6a21\u8fd0\u884c Airflow \u800c\u6784\u5efa\u7684\u89e3\u51b3\u65b9\u6848\u3002\u4f7f\u7528\u4e91\u7aef\u5b58\u50a8\u65f6\uff0c\u6587\u4ef6\u5b58\u53d6\u901f\u5ea6\u53ef\u80fd\u4f1a\u53d8\u6162<\/p>\n\n\n\n

\u5bf9\u4e8e Airflow \u73af\u5883\u7684\u6027\u80fd\u548c\u5b8c\u6574\u6027\uff0c\u5feb\u901f\u7684\u6587\u4ef6\u5b58\u53d6\u901f\u5ea6\u81f3\u5173\u91cd\u8981\u3002\u4e00\u4e2a\u6e05\u6670\u7684\u6587\u4ef6\u5b58\u53d6\u7b56\u7565\u53ef\u4ee5\u4fdd\u8bc1\u8c03\u5ea6\u5668\u80fd\u591f\u8fc5\u901f\u5730\u5bf9 DAG \u6587\u4ef6\u8fdb\u884c\u5904\u7406\uff0c\u5e76\u4e14\u8ba9\u4f60\u7684\u4f5c\u4e1a\u4fdd\u6301\u66f4\u65b0\u3002<\/p>\n\n\n\n

\u901a\u8fc7\u91cd\u590d\u626b\u63cf\u548c\u91cd\u65b0\u89e3\u6790\u914d\u7f6e\u7684 DAG \u76ee\u5f55\u4e2d\u7684\u6240\u6709\u6587\u4ef6\uff0c\u53ef\u4ee5\u4fdd\u6301\u5176\u5de5\u4f5c\u6d41\u7684\u5185\u90e8\u8868\u793a\u6700\u65b0\u3002\u8fd9\u4e9b\u6587\u4ef6\u5fc5\u987b\u7ecf\u5e38\u626b\u63cf\uff0c\u4ee5\u4fdd\u6301\u6bcf\u4e2a\u5de5\u4f5c\u8d1f\u8f7d\u7684\u78c1\u76d8\u6570\u636e\u6e90\u548c\u5176\u6570\u636e\u5e93\u5185\u90e8\u8868\u793a\u4e4b\u95f4\u7684\u4e00\u81f4\u6027\u3002\u8fd9\u5c31\u610f\u5473\u7740 DAG \u76ee\u5f55\u7684\u5185\u5bb9\u5fc5\u987b\u5728\u5355\u4e00\u73af\u5883\u4e2d\u7684\u6240\u6709\u8c03\u5ea6\u5668\u548c\u5de5\u4f5c\u5668\u4e4b\u95f4\u4fdd\u6301\u4e00\u81f4\uff08Airflow \u63d0\u4f9b\u4e86\u51e0\u79cd\u65b9\u6cd5\u6765\u5b9e\u73b0\u8fd9\u4e00\u76ee\u6807\uff09\u3002<\/p>\n\n\n\n

\u5728 Shopify \u4e2d\uff0c\u6211\u4eec\u5229\u7528\u8c37\u6b4c\u4e91\u5b58\u50a8\uff08Google Cloud Storage\uff0cGCS\uff09\u6765\u5b58\u50a8 DAG\u3002\u6211\u4eec\u6700\u521d\u90e8\u7f72 Airflow \u65f6\uff0c\u5229\u7528 GCSFuse \u5728\u5355\u4e00\u7684 Airflow \u73af\u5883\u4e2d\u7684\u6240\u6709\u5de5\u4f5c\u5668\u548c\u8c03\u5ea6\u5668\u6765\u7ef4\u62a4\u4e00\u81f4\u7684\u6587\u4ef6\u96c6\u3002\u7136\u800c\uff0c\u5728\u89c4\u6a21\u4e0a\uff0c\u8fd9\u88ab\u8bc1\u660e\u662f\u4e00\u4e2a\u6027\u80fd\u74f6\u9888\uff0c\u56e0\u4e3a\u6bcf\u4e2a\u6587\u4ef6\u7684\u5b58\u53d6\u90fd\u4f1a\u5f15\u8d77\u5bf9 GCS \u7684\u8bf7\u6c42\u3002\u7531\u4e8e\u5728\u73af\u5883\u4e2d\u7684\u6bcf\u4e00\u4e2a pod \u90fd\u9700\u8981\u5355\u72ec\u6302\u5728\u6876\uff0c\u6240\u4ee5\u5b58\u53d6\u91cf\u7279\u522b\u5927\u3002<\/p>\n\n\n\n

\u7ecf\u8fc7\u51e0\u6b21\u8bd5\u9a8c\uff0c\u6211\u4eec\u53d1\u73b0\uff0c\u5728 Kubernetes \u96c6\u7fa4\u4e0a\u8fd0\u884c\u4e00\u4e2a NFS\uff08Network file system\uff0c\u7f51\u7edc\u6587\u4ef6\u7cfb\u7edf\uff09\u670d\u52a1\u5668\uff0c\u53ef\u4ee5\u5927\u5927\u6539\u5584 Airflow \u73af\u5883\u7684\u6027\u80fd\u3002\u7136\u540e\uff0c\u6211\u4eec\u628a NFS \u670d\u52a1\u5668\u5f53\u4f5c\u4e00\u4e2a\u591a\u8bfb\u591a\u5199\u7684\u5377\u8f6c\u8fdb\u5de5\u4f5c\u5668\u548c\u8c03\u5ea6\u5668\u7684 pod \u4e2d\u3002\u6211\u4eec\u7f16\u5199\u4e86\u4e00\u4e2a\u81ea\u5b9a\u4e49\u811a\u672c\uff0c\u4f7f\u8be5\u5377\u7684\u72b6\u6001\u4e0e GCS \u540c\u6b65\uff0c\u56e0\u6b64\uff0c\u5f53 DAG \u88ab\u4e0a\u4f20\u6216\u8005\u7ba1\u7406\u65f6\uff0c\u7528\u6237\u53ef\u4ee5\u4e0e GCS \u8fdb\u884c\u4ea4\u4e92\u3002\u8fd9\u4e2a\u811a\u672c\u5728\u540c\u4e00\u4e2a\u96c6\u7fa4\u5185\u7684\u5355\u72ec pod \u4e2d\u8fd0\u884c\u3002\u8fd9\u4f7f\u5f97\u6211\u4eec\u53ef\u4ee5\u6709\u6761\u4ef6\u5730\u5728\u7ed9\u5b9a\u7684\u6876\u4e2d\u4ec5\u540c\u6b65 DAG \u7684\u5b50\u96c6\uff0c\u6216\u8005\u6839\u636e\u73af\u5883\u7684\u914d\u7f6e\uff0c\u5c06\u591a\u4e2a\u6876\u4e2d\u7684 DAG \u540c\u6b65\u5230\u4e00\u4e2a\u6587\u4ef6\u7cfb\u7edf\u4e2d\uff08\u7a0d\u540e\u4f1a\u8be6\u7ec6\u9610\u8ff0\uff09\u3002<\/p>\n\n\n\n

\u603b\u800c\u8a00\u4e4b\uff0c\u8fd9\u4e3a\u6211\u4eec\u63d0\u4f9b\u4e86\u5feb\u901f\u7684\u6587\u4ef6\u5b58\u53d6\u4f5c\u4e3a\u4e00\u4e2a\u7a33\u5b9a\u7684\u5916\u90e8\u6570\u636e\u6e90\uff0c\u540c\u65f6\u4fdd\u6301\u4e86\u6211\u4eec\u5feb\u901f\u6dfb\u52a0\u6216\u4fee\u6539 Airflow \u4e2d DAG \u6587\u4ef6\u7684\u80fd\u529b\u3002\u53e6\u5916\uff0c\u6211\u4eec\u8fd8\u53ef\u4ee5\u5229\u7528\u8c37\u6b4c\u4e91\u5e73\u53f0\u7684 IAM\uff08\u8bc6\u522b\u548c\u5b58\u53d6\u7ba1\u7406\uff09\u529f\u80fd\u6765\u63a7\u5236\u54ea\u4e9b\u7528\u6237\u80fd\u591f\u4e0a\u4f20\u6587\u4ef6\u5230\u7279\u5b9a\u7684\u73af\u5883\u3002\u4f8b\u5982\uff0c\u6211\u4eec\u53ef\u4ee5\u8ba9\u7528\u6237\u76f4\u63a5\u5c06 DAG \u76f4\u63a5\u4e0a\u4f20\u5230 staging \u73af\u5883\uff0c\u4f46\u5c06\u751f\u4ea7\u73af\u5883\u7684\u4e0a\u4f20\u9650\u5236\u5728\u6211\u4eec\u7684\u6301\u7eed\u90e8\u7f72\u8fc7\u7a0b\u4e2d\u3002<\/p>\n\n\n\n

\u5728\u5927\u89c4\u6a21\u8fd0\u884c Airflow \u65f6\uff0c\u786e\u4fdd\u5feb\u901f\u6587\u4ef6\u5b58\u53d6\u7684\u53e6\u4e00\u4e2a\u8003\u8651\u56e0\u7d20\u662f\u4f60\u7684\u6587\u4ef6\u5904\u7406\u6027\u80fd\u3002Airflow \u5177\u6709\u9ad8\u5ea6\u7684\u53ef\u914d\u7f6e\u6027\uff0c\u53ef\u4ee5\u901a\u8fc7\u591a\u79cd\u65b9\u6cd5\u8c03\u6574\u540e\u53f0\u6587\u4ef6\u5904\u7406\uff08\u4f8b\u5982\u6392\u5e8f\u6a21\u5f0f\u3001\u5e76\u884c\u6027\u548c\u8d85\u65f6\uff09\u3002\u8fd9\u4f7f\u5f97\u4f60\u53ef\u4ee5\u6839\u636e\u9700\u6c42\u4f18\u5316\u73af\u5883\uff0c\u4ee5\u5b9e\u73b0\u4ea4\u4e92\u5f0f DAG \u5f00\u53d1\u6216\u8c03\u5ea6\u5668\u6027\u80fd\u3002\u5143\u6570\u636e\u6570\u91cf\u7684\u589e\u52a0\uff0c\u53ef\u80fd\u4f1a\u964d\u4f4e Airflow \u8fd0\u884c\u6548\u7387<\/p>\n\n\n\n

\u5728\u4e00\u4e2a\u6b63\u5e38\u89c4\u6a21\u7684 Airflow \u90e8\u7f72\u4e2d\uff0c\u7531\u4e8e\u5143\u6570\u636e\u7684\u6570\u91cf\u800c\u9020\u6210\u7684\u6027\u80fd\u964d\u4f4e\u5e76\u4e0d\u662f\u95ee\u9898\uff0c\u81f3\u5c11\u5728\u6700\u521d\u7684\u51e0\u5e74\u91cc\u662f\u8fd9\u6837\u3002<\/p>\n\n\n\n

\u4f46\u662f\uff0c\u4ece\u89c4\u6a21\u4e0a\u770b\uff0c\u5143\u6570\u636e\u6b63\u5728\u8fc5\u901f\u5730\u7d2f\u79ef\u3002\u4e00\u6bb5\u65f6\u95f4\u4e4b\u540e\uff0c\u5c31\u53ef\u80fd\u5f00\u59cb\u5bf9\u6570\u636e\u5e93\u4ea7\u751f\u989d\u5916\u7684\u8d1f\u8f7d\u3002\u8fd9\u4e00\u70b9\u5728 Web \u7528\u6237\u754c\u9762\u7684\u52a0\u8f7d\u65f6\u95f4\u4e0a\u5c31\u53ef\u4ee5\u770b\u5f97\u51fa\u6765\uff0c\u5c24\u5176\u662f Airflow \u7684\u66f4\u65b0\uff0c\u5728\u8fd9\u6bb5\u65f6\u95f4\u91cc\uff0c\u8fc1\u79fb\u53ef\u80fd\u8981\u82b1\u8d39\u6570\u5c0f\u65f6\u3002<\/p>\n\n\n\n

\u7ecf\u8fc7\u53cd\u590d\u8bd5\u9a8c\uff0c\u6211\u4eec\u786e\u5b9a\u4e86 28 \u5929\u7684\u5143\u6570\u636e\u4fdd\u5b58\u7b56\u7565\uff0c\u5e76\u5b9e\u65bd\u4e86\u4e00\u4e2a\u7b80\u5355\u7684 DAG\uff0c\u5728 PythonOperator \u4e2d\u5229\u7528 ORM\uff08\u5bf9\u8c61\u5173\u7cfb\u6620\u5c04\uff09\u67e5\u8be2\uff0c\u4ece\u4efb\u4f55\u5305\u542b\u5386\u53f2\u6570\u636e\uff08DagRuns\u3001TaskInstances\u3001Logs\u3001TaskRetries \u7b49\uff09\u7684\u8868\u4e2d\u5220\u9664\u884c\u3002\u6211\u4eec\u4e4b\u6240\u4ee5\u9009\u62e9 28 \u5929\uff0c\u662f\u56e0\u4e3a\u5b83\u53ef\u4ee5\u8ba9\u6211\u4eec\u6709\u5145\u8db3\u7684\u5386\u53f2\u8bb0\u5f55\u6765\u7ba1\u7406\u4e8b\u4ef6\u548c\u8ddf\u8e2a\u5386\u53f2\u5de5\u4f5c\u7ee9\u6548\uff0c\u540c\u65f6\u5c06\u6570\u636e\u5e93\u4e2d\u7684\u6570\u636e\u91cf\u4fdd\u6301\u5728\u5408\u7406\u7684\u6c34\u5e73\u3002<\/p>\n\n\n\n

db_clean.py\uff1a<\/p>\n\n\n\n

import logging\r\nfrom datetime import datetime, timezone, timedelta\r\n\r\nfrom sqlalchemy import delete\r\nfrom airflow.models import DAG, Log, DagRun, TaskInstance, TaskReschedule, Variable\r\nfrom airflow.jobs.base_job import BaseJob\r\nfrom airflow.utils.dates import days_ago\r\nfrom airflow.operators.python import PythonOperator\r\nfrom airflow.utils.state import State\r\nfrom airflow.utils.session import provide_session\r\n\r\nEXPIRATION_WEEKS = 4\r\n\r\n@provide_session\r\ndef delete_old_database_entries_by_model(table, date_col, session=None):\r\n    expiration_date = datetime.now(timezone.utc) - timedelta(weeks=EXPIRATION_WEEKS)\r\n    query = delete(table).where(date_col < expiration_date)\r\n\r\n    if \"state\" in dir(table):\r\n        query = query.where(State.RUNNING != \"state\")\r\n\r\n    result = session.execute(query)\r\n    logging.info(\r\n        \"Deleted %s rows from the database for the %s table that are older than %s.\",\r\n        result.rowcount,\r\n        table,\r\n        expiration_date,\r\n    )\r\n\r\ndef delete_old_database_entries():\r\n\r\n    if Variable.get(\"ENABLE_DB_TRUNCATION\", \"\") != \"True\":\r\n        logging.warning(\"This DAG will delete all data older than %s weeks.\", EXPIRATION_WEEKS)\r\n        logging.warning(\"To enable this, create an Airflow Variable called ENABLE_DB_TRUNCATION set to 'True'\")\r\n        logging.warning(\"Skipping truncation until explicitly enabled.\")\r\n        return\r\n\r\n    delete_old_database_entries_by_model(TaskInstance, TaskInstance.end_date)\r\n    delete_old_database_entries_by_model(DagRun, DagRun.end_date)\r\n    delete_old_database_entries_by_model(BaseJob, BaseJob.end_date)\r\n    delete_old_database_entries_by_model(Log, Log.dttm)\r\n    delete_old_database_entries_by_model(TaskReschedule, TaskReschedule.end_date)\r\n\r\ndag = DAG(\r\n    \"airflow-utils.truncate-database\",\r\n    start_date=days_ago(1),\r\n    max_active_runs=1,\r\n    dagrun_timeout=timedelta(minutes=20),\r\n    schedule_interval=\"@daily\",\r\n    catchup=False,\r\n)\r\n\r\nPythonOperator(\r\n    task_id=\"cleanup-old-database-entries\",\r\n    dag=dag,\r\n    python_callable=delete_old_database_entries,\r\n)<\/code><\/pre>\n\n\n\n
<\/pre>\n\n\n\n
\u9057\u61be\u7684\u662f\uff0c\u8fd9\u5c31\u610f\u5473\u7740\uff0c\u5728\u6211\u4eec\u7684\u73af\u5883\u4e2d\uff0cAirflow \u4e2d\u7684\u90a3\u4e9b\u4f9d\u8d56\u4e8e\u6301\u4e45\u4f5c\u4e1a\u5386\u53f2\u7684\u7279\u6027\uff08\u4f8b\u5982\uff0c\u957f\u65f6\u95f4\u7684\u56de\u586b\uff09\u5e76\u4e0d\u88ab\u652f\u6301\u3002\u8fd9\u5bf9\u6211\u4eec\u6765\u8bf4\u5e76\u4e0d\u662f\u4e00\u4e2a\u95ee\u9898\uff0c\u4f46\u662f\u5b83\u6709\u53ef\u80fd\u4f1a\u5bfc\u81f4\u95ee\u9898\uff0c\u8fd9\u8981\u53d6\u51b3\u4e8e\u4f60\u7684\u4fdd\u5b58\u671f\u548c Airflow \u7684\u4f7f\u7528\u60c5\u51b5\u3002<\/p>\n\n\n\n
\u4f5c\u4e3a\u81ea\u5b9a\u4e49 DAG \u7684\u53e6\u4e00\u79cd\u65b9\u6cd5\uff0cAirflow \u6700\u8fd1\u589e\u52a0\u4e86\u5bf9 db clean \u547d\u4ee4\u7684\u652f\u6301\uff0c\u53ef\u4ee5\u7528\u6765\u5220\u9664\u65e7\u7684\u5143\u6570\u636e\u3002\u8fd9\u4e2a\u547d\u4ee4\u5728 Airflow 2.3 \u7248\u672c\u4e2d\u53ef\u7528\u3002DAG \u53ef\u80fd\u5f88\u96be\u4e0e\u7528\u6237\u548c\u56e2\u961f\u5173\u8054<\/p>\n\n\n\n
\u5728\u591a\u79df\u6237\u73af\u5883\u4e2d\u8fd0\u884c Airflow \u65f6\uff08\u5c24\u5176\u662f\u5728\u5927\u578b\u7ec4\u7ec7\u4e2d\uff09\uff0c\u80fd\u591f\u5c06 DAG \u8ffd\u6eaf\u5230\u4e2a\u4eba\u6216\u56e2\u961f\u662f\u5f88\u91cd\u8981\u7684\u3002\u4e3a\u4ec0\u4e48\uff1f\u56e0\u4e3a\u5982\u679c\u4e00\u4e2a\u4f5c\u4e1a\u5931\u8d25\u4e86\uff0c\u629b\u51fa\u9519\u8bef\u6216\u5e72\u6270\u5176\u4ed6\u5de5\u4f5c\u8d1f\u8f7d\uff0c\u6211\u4eec\u7684\u7ba1\u7406\u5458\u53ef\u4ee5\u8fc5\u901f\u8054\u7cfb\u5230\u5408\u9002\u7684\u7528\u6237\u3002<\/p>\n\n\n\n
\u5982\u679c\u6240\u6709\u7684 DAG \u90fd\u76f4\u63a5\u4ece\u4e00\u4e2a\u4ed3\u5e93\u90e8\u7f72\uff0c\u6211\u4eec\u53ef\u4ee5\u7b80\u5355\u5730\u4f7f\u7528 git blame \u6765\u8ffd\u8e2a\u5de5\u4f5c\u7684\u6240\u6709\u8005\u3002\u7136\u800c\uff0c\u7531\u4e8e\u6211\u4eec\u5141\u8bb8\u7528\u6237\u4ece\u81ea\u5df1\u7684\u9879\u76ee\u4e2d\u90e8\u7f72\u5de5\u4f5c\u8d1f\u8f7d\uff08\u751a\u81f3\u5728\u90e8\u7f72\u65f6\u52a8\u6001\u751f\u6210\u4f5c\u4e1a\uff09\uff0c\u8fd9\u5c31\u53d8\u5f97\u66f4\u52a0\u56f0\u96be\u3002<\/p>\n\n\n\n
\u4e3a\u4e86\u65b9\u4fbf\u8ffd\u8e2a DAG \u7684\u6765\u6e90\uff0c\u6211\u4eec\u5f15\u5165\u4e86\u4e00\u4e2a Airflow \u547d\u540d\u7a7a\u95f4\u7684\u6ce8\u518c\u8868\uff0c\u5e76\u5c06\u5176\u79f0\u4e3a Airflow \u73af\u5883\u7684\u6e05\u5355\u6587\u4ef6\u3002<\/p>\n\n\n\n
sample_airflow_manifest.yaml\uff1a<\/p>\n\n\n\n
projects:\r\n  defaults: &defaults\r\n    source_repository: 'https:\/\/github.com\/my_organization\/dag_repo'\r\n    dag_source_bucket: 'my_organization_dags'\r\n    constraints: &constraints\r\n      airflow_celery_queues:\r\n        - 'default'\r\n      pools:\r\n        - 'default'\r\n  data_extracts:\r\n    << : *defaults\r\n    owner_email: 'etl-team@my-organization.com'\r\n    source_repository: 'https:\/\/github.com\/my_organization\/airflow_extracts'\r\n    constraints:\r\n      << : *constraints\r\n      namespaces:\r\n        - 'etl-jobs'\r\n      pools:\r\n        - 'extracts'\r\n  batch_processing:\r\n    <<: *defaults\r\n    owner_email: 'spark-team@my-organization.com'\r\n    source_repository: 'https:\/\/github.com\/Shopify\/airflow_batch_jobs'\r\n    constraints:\r\n      <<: *constraints<\/code><\/pre>\n\n\n\n
<\/pre>\n\n\n\n
\u6e05\u5355\u6587\u4ef6\u662f\u4e00\u4e2a YAML \u6587\u4ef6\uff0c\u7528\u6237\u5fc5\u987b\u4e3a\u4ed6\u4eec\u7684 DAG \u6ce8\u518c\u4e00\u4e2a\u547d\u540d\u7a7a\u95f4\u3002\u5728\u8fd9\u4e2a\u6587\u4ef6\u4e2d\uff0c\u4ed6\u4eec\u5c06\u5305\u62ec\u4f5c\u4e1a\u7684\u6240\u6709\u8005\u548c\u6e90 github \u4ed3\u5e93\uff08\u751a\u81f3\u662f\u6e90 GCS \u6876\uff09\u7684\u4fe1\u606f\uff0c\u4ee5\u53ca\u4e3a\u5176 DAG \u5b9a\u4e49\u4e00\u4e9b\u57fa\u672c\u9650\u5236\u3002\u6211\u4eec\u4e3a\u6bcf\u4e2a\u73af\u5883\u7ef4\u62a4\u4e00\u4e2a\u5355\u72ec\u7684\u6e05\u5355\uff0c\u5e76\u5c06\u5176\u4e0e DAG \u4e00\u8d77\u4e0a\u4f20\u5230 GCS\u3002DAG \u4f5c\u8005\u6709\u5f88\u5927\u7684\u6743\u529b<\/p>\n\n\n\n
\u901a\u8fc7\u5141\u8bb8\u7528\u6237\u76f4\u63a5\u7f16\u5199\u548c\u4e0a\u4f20 DAG \u5230\u5171\u4eab\u73af\u5883\uff0c\u6211\u4eec\u8d4b\u4e88\u4e86\u4ed6\u4eec\u5f88\u5927\u7684\u6743\u529b\u3002\u7531\u4e8e Airflow \u662f\u6211\u4eec\u6570\u636e\u5e73\u53f0\u7684\u6838\u5fc3\u7ec4\u6210\u90e8\u5206\uff0c\u5b83\u4e0e\u8bb8\u591a\u4e0d\u540c\u7684\u7cfb\u7edf\u76f8\u8054\u7cfb\uff0c\u56e0\u6b64\u4f5c\u4e1a\u6709\u5e7f\u6cdb\u7684\u8bbf\u95ee\u6743\u3002\u867d\u7136\u6211\u4eec\u4fe1\u4efb\u6211\u4eec\u7684\u7528\u6237\uff0c\u4f46\u6211\u4eec\u4ecd\u7136\u5e0c\u671b\u5bf9\u4ed6\u4eec\u5728\u7279\u5b9a\u7684 Airflow \u73af\u5883\u4e2d\u80fd\u505a\u4ec0\u4e48\u548c\u4e0d\u80fd\u505a\u4ec0\u4e48\u4fdd\u6301\u4e00\u5b9a\u7a0b\u5ea6\u7684\u63a7\u5236\u3002\u8fd9\u4e00\u70b9\u5728\u89c4\u6a21\u4e0a\u5c24\u4e3a\u91cd\u8981\uff0c\u56e0\u4e3a\u8981\u8ba9 Airflow \u7ba1\u7406\u5458\u5728\u6240\u6709\u4f5c\u4e1a\u8fdb\u5165\u751f\u4ea7\u4e4b\u524d\u5bf9\u5176\u8fdb\u884c\u5ba1\u67e5\u662f\u4e0d\u73b0\u5b9e\u7684\u3002<\/p>\n\n\n\n
\u4e3a\u4e86\u521b\u5efa\u4e00\u4e9b\u57fa\u672c\u7684\u201c\u62a4\u680f\u201d\uff0c\u6211\u4eec\u91c7\u7528\u4e86\u4e00\u4e2a DAG \u7b56\u7565\uff0c\u5b83\u4ece\u4e4b\u524d\u63d0\u5230\u7684 Airflow \u6e05\u5355\u4e2d\u8bfb\u53d6\u914d\u7f6e\uff0c\u5e76\u901a\u8fc7\u5f15\u53d1 AirflowClusterPolicyViolation \u6765\u62d2\u7edd\u90a3\u4e9b\u4e0d\u7b26\u5408\u5176\u547d\u540d\u7a7a\u95f4\u7ea6\u675f\u7684 DAG\u3002<\/p>\n\n\n\n
\u6839\u636e\u6e05\u5355\u6587\u4ef6\u7684\u5185\u5bb9\uff0c\u8be5\u7b56\u7565\u5c06\u5bf9 DAG \u6587\u4ef6\u5e94\u7528\u4e00\u4e9b\u57fa\u672c\u9650\u5236\uff0c\u4f8b\u5982\uff1a<\/p>\n\n\n\n
    \n
  • DAG ID \u5fc5\u987b\u4ee5\u73b0\u6709\u540d\u79f0\u7a7a\u95f4\u7684\u540d\u79f0\u4e3a\u524d\u7f00<\/strong>\uff0c\u4ee5\u83b7\u5f97\u6240\u6709\u6743\u3002<\/li>\n\n\n\n
  • DAG \u4e2d\u7684\u4efb\u52a1\u5fc5\u987b\u53ea\u5411\u6307\u5b9a\u7684 celery \u961f\u5217\u53d1\u51fa\u4efb\u52a1<\/strong>\uff0c\u8fd9\u4e2a\u5c06\u5728\u540e\u9762\u8ba8\u8bba\u3002<\/li>\n\n\n\n
  • DAG \u4e2d\u7684\u4efb\u52a1\u53ea\u80fd\u5728\u6307\u5b9a\u7684\u6c60\u4e2d\u8fd0\u884c<\/strong>\uff0c\u4ee5\u9632\u6b62\u4e00\u4e2a\u5de5\u4f5c\u8d1f\u8f7d\u5360\u7528\u53e6\u4e00\u4e2a\u7684\u5bb9\u91cf\u3002<\/li>\n\n\n\n
  • \u8be5 DAG \u4e2d\u7684\u4efb\u4f55<\/strong> KubernetesPodOperators \u5fc5\u987b\u53ea\u5728\u6307\u5b9a\u7684\u547d\u540d\u7a7a\u95f4\u4e2d\u542f\u52a8 pod<\/strong>\uff0c\u4ee5\u9632\u6b62\u5b58\u53d6\u5176\u4ed6\u547d\u540d\u7a7a\u95f4\u7684\u79d8\u5bc6\u3002<\/li>\n\n\n\n
  • DAG \u4e2d\u7684\u4efb\u52a1\u53ea\u80fd\u5411\u6307\u5b9a\u7684\u5916\u90e8 kubernetes \u96c6\u7fa4\u96c6\u53d1\u5c04 pod\u3002<\/strong>\u8fd9\u4e2a\u7b56\u7565\u8fd8\u53ef\u4ee5\u5ef6\u4f38\u5230\u6267\u884c\u5176\u4ed6\u89c4\u5219\uff08\u4f8b\u5982\uff0c\u53ea\u5141\u8bb8\u4e00\u7ec4\u6709\u9650\u7684\u64cd\u4f5c\u8005\uff09\uff0c\u751a\u81f3\u53ef\u4ee5\u5c06\u4efb\u52a1\u8fdb\u884c\u7a81\u53d8\uff0c\u4ee5\u6ee1\u8db3\u67d0\u79cd\u89c4\u8303\uff08\u4f8b\u5982\uff0c\u4e3a DAG \u4e2d\u7684\u6240\u6709\u4efb\u52a1\u6dfb\u52a0\u4e00\u4e2a\u7279\u5b9a\u547d\u540d\u7a7a\u95f4\u7684\u6267\u884c\u8d85\u65f6\uff09\u3002<\/li>\n<\/ul>\n\n\n\n
    \u4e0b\u9762\u662f\u4e00\u4e2a\u7b80\u5316\u7684\u4f8b\u5b50\uff0c\u6f14\u793a\u5982\u4f55\u521b\u5efa\u4e00\u4e2a DAG \u7b56\u7565\uff0c\u8be5\u7b56\u7565\u8bfb\u53d6\u5148\u524d\u5171\u4eab\u7684\u6e05\u5355\u6587\u4ef6\uff0c\u5e76\u5b9e\u73b0\u4e0a\u8ff0\u524d\u4e09\u9879\u63a7\u5236\uff1a<\/p>\n\n\n\n
    airflow_local_settings.py\uff1a<\/p>\n\n\n\n
    import os\r\nfrom typing import List\r\nimport yaml\r\nfrom airflow.exceptions import AirflowClusterPolicyViolation\r\nfrom airflow.models import DAG\r\n\r\ndef validate_pools(dag: DAG, pools: List[str]) -> None:\r\n    for task in dag.tasks:\r\n        if task.pool not in pools:\r\n            raise AirflowClusterPolicyViolation(\r\n                f\"DAG {dag.dag_id} cannot submit tasks to the pool: {task.pool}\"\r\n            )\r\n\r\ndef validate_queues(dag: DAG, queues: List[str]) -> None:\r\n    for task in dag.tasks:\r\n        if task.queue not in queues:\r\n            raise AirflowClusterPolicyViolation(\r\n                f\"DAG {dag.dag_id} cannot submit tasks to the queue: {task.queue}\"\r\n            )\r\n\r\ndef dag_policy(dag: DAG) -> None:\r\n\r\n    airflow_home = os.environ.get('AIRFLOW_HOME', '~\/airflow')\r\n    manifest_path = f\"{airflow_home}\/airflow_manifest.yaml\"\r\n    with open(manifest_path, \"r\", encoding=\"UTF-8\") as manifest_file:\r\n        manifest = yaml.safe_load(manifest_file)\r\n\r\n    dag_namespace = dag.dag_id.split(\".\")[0]\r\n    if dag_namespace not in manifest[\"projects\"]:\r\n        raise AirflowClusterPolicyViolation(\r\n            f\"Namespace {dag_namespace} is not registered in the Airflow Manifest.\"\r\n        )\r\n\r\n    constraints = manifest[\"projects\"][dag_namespace][\"constraints\"]\r\n\r\n    validate_pools(dag, constraints[\"pools\"])\r\n    validate_queues(dag, constraints[\"queues\"])<\/code><\/pre>\n\n\n\n
    \u8fd9\u4e9b\u9a8c\u8bc1\u4e3a\u6211\u4eec\u63d0\u4f9b\u4e86\u8db3\u591f\u7684\u53ef\u8ffd\u6eaf\u6027\uff0c\u540c\u65f6\u4e5f\u521b\u9020\u4e86\u4e00\u4e9b\u57fa\u672c\u7684\u63a7\u5236\uff0c\u4ece\u800c\u51cf\u5c11\u4e86 DAG \u4e4b\u95f4\u7684\u76f8\u4e92\u5e72\u6270\u80fd\u529b\u3002\u5f88\u96be\u786e\u4fdd\u8d1f\u8f7d\u7684\u4e00\u81f4\u5206\u5e03<\/p>\n\n\n\n
    \u5bf9\u4f60\u7684 DAG \u7684\u8ba1\u5212\u95f4\u9694\u4e2d\u4f7f\u7528\u4e00\u4e2a\u7edd\u5bf9\u7684\u95f4\u9694\u662f\u5f88\u6709\u5438\u5f15\u529b\u7684\uff1a\u7b80\u5355\u5730\u8bbe\u7f6e DAG \u6bcf\u8fd0\u884c\u4e00\u6b21 timedelta(hours=1)\uff0c\u4f60\u5c31\u53ef\u4ee5\u653e\u5fc3\u5730\u79bb\u5f00\uff0c\u56e0\u4e3a\u4f60\u77e5\u9053 DAG \u5c06\u5927\u7ea6\u6bcf\u5c0f\u65f6\u8fd0\u884c\u4e00\u6b21\u3002\u7136\u800c\uff0c\u8fd9\u53ef\u80fd\u4f1a\u5bfc\u81f4\u89c4\u6a21\u4e0a\u7684\u95ee\u9898\u3002<\/p>\n\n\n\n
    \u5f53\u7528\u6237\u5408\u5e76\u5927\u91cf\u81ea\u52a8\u751f\u6210\u7684 DAG\uff0c\u6216\u8005\u7f16\u5199\u4e00\u4e2a Python \u6587\u4ef6\uff0c\u5728\u89e3\u6790\u65f6\u751f\u6210\u8bb8\u591a DAG\uff0c\u6240\u6709\u7684 DAGRuns \u5c06\u5728\u540c\u4e00\u65f6\u95f4\u88ab\u521b\u5efa\u3002\u8fd9\u4f1a\u5bfc\u81f4\u5927\u91cf\u7684\u6d41\u91cf\uff0c\u4f7f Airflow \u8c03\u5ea6\u5668\u4ee5\u53ca\u4f5c\u4e1a\u6240\u4f7f\u7528\u7684\u4efb\u4f55\u5916\u90e8\u670d\u52a1\u6216\u57fa\u7840\u8bbe\u65bd\u8d85\u8f7d\uff0c\u6bd4\u5982 Trino \u96c6\u7fa4\u3002<\/p>\n\n\n\n
    \u5728\u4e00\u4e2a schedule_interval \u901a\u8fc7\u4e4b\u540e\uff0c\u6240\u6709\u8fd9\u4e9b\u4f5c\u4e1a\u5c06\u5728\u540c\u4e00\u65f6\u95f4\u518d\u6b21\u8fd0\u884c\uff0c\u4ece\u800c\u5bfc\u81f4\u53e6\u4e00\u4e2a\u6d41\u91cf\u6fc0\u589e\u3002\u6700\u7ec8\uff0c\u8fd9\u53ef\u80fd\u5bfc\u81f4\u8d44\u6e90\u5229\u7528\u7387\u4e0d\u7406\u60f3\uff0c\u6267\u884c\u65f6\u95f4\u589e\u52a0\u3002<\/p>\n\n\n\n
    \u867d\u7136\u57fa\u4e8e crontab \u7684\u65f6\u95f4\u8868\u4e0d\u4f1a\u5bfc\u81f4\u8fd9\u79cd\u6fc0\u589e\uff0c\u4f46\u5b83\u4eec\u4e5f\u5b58\u5728\u81ea\u5df1\u7684\u95ee\u9898\u3002\u4eba\u7c7b\u504f\u5411\u4e8e\u4eba\u7c7b\u53ef\u8bfb\u7684\u65f6\u95f4\u8868\uff0c\u56e0\u6b64\u503e\u5411\u4e8e\u521b\u5efa\u5728\u6574\u70b9\u3001\u6bcf\u5c0f\u65f6\u3001\u6bcf\u665a\u7684\u5348\u591c\u8fd0\u884c\u7684\u4f5c\u4e1a\uff0c\u7b49\u7b49\u3002\u6709\u65f6\u5019\uff0c\u5b83\u53ef\u4ee5\u4e3a\u67d0\u4e00\u7279\u5b9a\u7684\u5e94\u7528\u63d0\u4f9b\u4e00\u4e2a\u5408\u7406\u7684\u7406\u7531\uff08\u6bd4\u5982\uff0c\u6211\u4eec\u5e0c\u671b\u5728\u6bcf\u4e2a\u665a\u4e0a\u534a\u591c\u6536\u96c6\u524d\u4e00\u5929\u7684\u6570\u636e\uff09\uff0c\u4f46\u662f\u6211\u4eec\u5e38\u5e38\u4f1a\u53d1\u73b0\uff0c\u7528\u6237\u4ec5\u4ec5\u5e0c\u671b\u5728\u4e00\u4e2a\u56fa\u5b9a\u7684\u65f6\u95f4\u95f4\u9694\u5185\u8fd0\u884c\u4ed6\u4eec\u7684\u4f5c\u4e1a\u3002\u5982\u679c\u7528\u6237\u53ef\u4ee5\u76f4\u63a5\u6307\u5b9a\u81ea\u5df1\u7684 crontab\uff0c\u90a3\u4e48\u5c06\u4f1a\u9020\u6210\u6d41\u91cf\u7684\u6fc0\u589e\uff0c\u8fd9\u5c06\u4f1a\u5bf9 SLO \u9020\u6210\u5f71\u54cd\uff0c\u5e76\u4e14\u4f7f\u5916\u90e8\u7cfb\u7edf\u7684\u8d1f\u8f7d\u4e0d\u5e73\u8861\u3002<\/p>\n\n\n\n
    \u4f5c\u4e3a\u8fd9\u4e24\u4e2a\u95ee\u9898\u7684\u89e3\u51b3\u65b9\u6848\uff0c\u6211\u4eec\u5bf9\u6240\u6709\u81ea\u52a8\u751f\u6210\u7684 DAG\uff08\u4ee3\u8868\u4e86\u6211\u4eec\u7edd\u5927\u591a\u6570\u7684\u5de5\u4f5c\u6d41\uff09\u4f7f\u7528\u4e00\u4e2a\u786e\u5b9a\u6027\u7684\u968f\u673a\u65f6\u95f4\u8868\u95f4\u9694\u3002\u8fd9\u901a\u5e38\u662f\u57fa\u4e8e\u4e00\u4e2a\u6052\u5b9a\u79cd\u5b50\u7684\u54c8\u5e0c\u503c\uff0c\u5982 dag_id\u3002<\/p>\n\n\n\n
    \u4e0b\u9762\u7684\u7247\u6bb5\u63d0\u4f9b\u4e86\u4e00\u4e2a\u7b80\u5355\u7684\u51fd\u6570\u793a\u4f8b\uff0c\u8be5\u51fd\u6570\u751f\u6210\u786e\u5b9a\u6027\u7684\u3001\u968f\u673a\u7684 crontab\uff0c\u4ea7\u751f\u6052\u5b9a\u7684\u65f6\u95f4\u8868\u95f4\u9694\u3002\u9057\u61be\u7684\u662f\uff0c\u7531\u4e8e\u5e76\u975e\u5168\u90e8\u95f4\u9694\u90fd\u53ef\u4ee5\u7528 crontab \u8868\u793a\uff0c\u56e0\u6b64\u5b83\u4f1a\u9650\u5236\u53ef\u80fd\u7684\u95f4\u9694\u8303\u56f4\u3002\u6211\u4eec\u5e76\u6ca1\u6709\u53d1\u73b0\u8fd9\u79cd\u6709\u9650\u7684\u65f6\u95f4\u8868\u95f4\u9694\u7684\u9009\u62e9\u662f\u6709\u5c40\u9650\u6027\u7684\uff0c\u5728\u6211\u4eec\u786e\u5b9e\u9700\u8981\u6bcf\u4e94\u5c0f\u65f6\u8fd0\u884c\u4e00\u4e2a\u4f5c\u4e1a\u7684\u60c5\u51b5\u4e0b\uff0c\u6211\u4eec\u53ea\u662f\u63a5\u53d7\u6bcf\u5929\u4f1a\u6709\u4e00\u4e2a\u56db\u5c0f\u65f6\u7684\u95f4\u9694\u3002<\/p>\n\n\n\n
    random_airflow_schedule.py\uff1a\nfrom hashlib import md5\nfrom random import randint, seed\n\ndef compute_schedule(dag_id, interval):\n\n    # create deterministic randomness by seeding PRNG with a hash of the table name:\n    seed(md5(dag_id.encode()).hexdigest())\n\n    if interval.endswith(\"h\"):\n        val = int(interval[:-1])\n        if 24 % val != 0:\n            raise ValueError(\"Must use a number which evenly divides 24.\")\n        offset = randint(0, val-1)\n        minutes = randint(0, 59)\n        return f\"{minutes} {offset}\/{val} * * *\"\n\n    elif interval.endswith(\"m\"):\n        val = int(interval[:-1])\n        if 60 % val != 0:\n            raise ValueError(\"Minutes must use a number which evenly divides 60.\")\n        offset = randint(0, val-1)\n        return f\"{offset}\/{val} * * * *\"\n\n    elif interval == \"1d\":\n        return f\"{randint(0, 59)} {randint(0, 23)} * * *\"\n\n    raise ValueError(\"Interval must be (1, 2, 3, 4, 5, 6, 10, 12, 15, 20, 30)m, (1, 2, 3, 4, 6, 12)h or 1d\")\n<\/pre>\n\n\n\n
    \u5f97\u76ca\u4e8e\u6211\u4eec\u7684\u968f\u673a\u65f6\u95f4\u8868\u7684\u5b9e\u65bd\uff0c\u6211\u4eec\u80fd\u591f\u5927\u5927\u5e73\u6ed1\u8d1f\u8f7d\u3002\u4e0b\u56fe\u663e\u793a\u4e86\u5728\u6211\u4eec\u6700\u5927\u7684\u5355\u4e00 Airflow \u73af\u5883\u4e2d\uff0c\u6bcf 10 \u5206\u949f\u5b8c\u6210\u7684\u4efb\u52a1\u6570\u3002<\/p>\n\n\n\n
    <\/p>\n\n\n\n
    \u5728\u6211\u4eec\u7684\u751f\u4ea7 Airflow \u73af\u5883\u4e2d\uff0c\u6bcf 10 \u5206\u949f\u6267\u884c\u4e00\u6b21\u4efb\u52a1<\/code><\/code><\/pre>\n\n\n\n
    \u5b58\u5728\u8bb8\u591a\u8d44\u6e90\u4e89\u7528\u70b9<\/pre>\n\n\n\n
    \"\"<\/figure>\n\n\n\n
    \u5728 Airflow \u4e2d\uff0c\u5b58\u5728\u7740\u5f88\u591a\u53ef\u80fd\u7684\u8d44\u6e90\u4e89\u7528\u70b9\uff0c\u901a\u8fc7\u4e00\u7cfb\u5217\u5b9e\u9a8c\u6027\u7684\u914d\u7f6e\u6539\u53d8\uff0c\u6700\u7ec8\u5f88\u5bb9\u6613\u51fa\u73b0\u74f6\u9888\u95ee\u9898\u3002\u5176\u4e2d\u4e00\u4e9b\u8d44\u6e90\u51b2\u7a81\u53ef\u4ee5\u5728 Airflow \u5185\u90e8\u5904\u7406\uff0c\u800c\u53e6\u4e00\u4e9b\u53ef\u80fd\u9700\u8981\u4e00\u4e9b\u57fa\u7840\u8bbe\u65bd\u7684\u6539\u53d8\u3002\u4ee5\u4e0b\u662f\u6211\u4eec\u5728 Shopify \u7684 Airflow \u4e2d\u5904\u7406\u8d44\u6e90\u4e89\u7528\u7684\u51e0\u79cd\u65b9\u6cd5\uff1a\u6c60<\/p>\n\n\n\n
    \"\"\/<\/figure>\n\n\n\n
    \u51cf\u5c11\u8d44\u6e90\u4e89\u7528\u7684\u4e00\u79cd\u65b9\u6cd5\u662f\u4f7f\u7528 Airflow \u6c60\u3002\u6c60\u7528\u4e8e\u9650\u5236\u4e00\u7ec4\u7279\u5b9a\u4efb\u52a1\u7684\u5e76\u53d1\u6027\u3002\u8fd9\u5bf9\u4e8e\u51cf\u5c11\u6d41\u91cf\u6fc0\u589e\u5f15\u8d77\u7684\u4e2d\u65ad\u975e\u5e38\u6709\u7528\u3002\u867d\u7136\u6c60\u662f\u6267\u884c\u4efb\u52a1\u9694\u79bb\u7684\u6709\u7528\u5de5\u5177\uff0c\u4f46\u7531\u4e8e\u53ea\u6709\u7ba1\u7406\u5458\u53ef\u4ee5\u901a\u8fc7 Web UI \u7f16\u8f91\u6c60\uff0c\u56e0\u6b64\u5728\u7ba1\u7406\u4e0a\u662f\u4e00\u4e2a\u6311\u6218\u3002<\/p>\n\n\n\n
    \u6211\u4eec\u7f16\u5199\u4e86\u4e00\u4e2a\u81ea\u5b9a\u4e49\u7684 DAG\uff0c\u901a\u8fc7\u4e00\u4e9b\u7b80\u5355\u7684 ORM \u67e5\u8be2\uff0c\u5c06\u6211\u4eec\u73af\u5883\u4e2d\u7684\u6c60\u4e0e Kubernetes Configmao \u4e2d\u6307\u5b9a\u7684\u72b6\u6001\u540c\u6b65\u3002\u8fd9\u8ba9\u6211\u4eec\u53ef\u4ee5\u5728\u7ba1\u7406 Airflow \u90e8\u7f72\u914d\u7f6e\u7684\u540c\u65f6\u7ba1\u7406\u6c60\uff0c\u5e76\u5141\u8bb8\u7528\u6237\u901a\u8fc7\u5ba1\u67e5\u7684\u62c9\u53d6\u8bf7\u6c42\u6765\u66f4\u65b0\u6c60\uff0c\u800c\u4e0d\u9700\u8981\u63d0\u5347\u8bbf\u95ee\u6743\u9650\u3002\u4f18\u5148\u7ea7\u6743\u91cd<\/p>\n\n\n\n
    Priority_weight \u5141\u8bb8\u4f60\u4e3a\u4e00\u4e2a\u7ed9\u5b9a\u7684\u4efb\u52a1\u5206\u914d\u4e00\u4e2a\u66f4\u9ad8\u7684\u4f18\u5148\u7ea7\u3002\u5177\u6709\u8f83\u9ad8\u4f18\u5148\u7ea7\u7684\u4efb\u52a1\u5c06\u201c\u6d6e\u52a8\u201d\u5230\u5806\u7684\u9876\u90e8\uff0c\u88ab\u9996\u5148\u5b89\u6392\u3002\u867d\u7136\u4e0d\u662f\u8d44\u6e90\u4e89\u7528\u7684\u76f4\u63a5\u89e3\u51b3\u65b9\u6848\uff0c\u4f46 priority_weight \u5bf9\u4e8e\u786e\u4fdd\u5ef6\u8fdf\u654f\u611f\u7684\u5173\u952e\u4efb\u52a1\u5728\u4f4e\u4f18\u5148\u7ea7\u4efb\u52a1\u4e4b\u524d\u8fd0\u884c\u662f\u5f88\u6709\u7528\u7684\u3002\u7136\u800c\uff0c\u9274\u4e8e priority_weight \u662f\u4e00\u4e2a\u4efb\u610f\u7684\u5c3a\u5ea6\uff0c\u5982\u679c\u4e0d\u4e0e\u6240\u6709\u5176\u4ed6\u4efb\u52a1\u8fdb\u884c\u6bd4\u8f83\uff0c\u5c31\u5f88\u96be\u786e\u5b9a\u4e00\u4e2a\u4efb\u52a1\u7684\u5b9e\u9645\u4f18\u5148\u7ea7\u3002\u6211\u4eec\u7528\u5b83\u6765\u786e\u4fdd\u6211\u4eec\u7684\u57fa\u672c Airflow \u76d1\u63a7 DAG\uff08\u5b83\u53d1\u51fa\u7b80\u5355\u7684\u6307\u6807\u5e76\u4e3a\u4e00\u4e9b\u8b66\u62a5\u63d0\u4f9b\u52a8\u529b\uff09\u603b\u662f\u5c3d\u53ef\u80fd\u53ca\u65f6\u5730\u8fd0\u884c\u3002<\/p>\n\n\n\n
    \u540c\u6837\u503c\u5f97\u6ce8\u610f\u7684\u662f\uff0c\u5728\u9ed8\u8ba4\u60c5\u51b5\u4e0b\uff0c\u4e00\u4e2a\u4efb\u52a1\u5728\u505a\u8c03\u5ea6\u51b3\u7b56\u65f6\u4f7f\u7528\u7684\u6709\u6548 priority_weight \u662f\u5176\u81ea\u8eab\u548c\u6240\u6709\u4e0b\u6e38\u4efb\u52a1\u7684\u6743\u91cd\u4e4b\u548c\u3002\u8fd9\u610f\u5473\u7740\uff0c\u5927 DAG \u4e2d\u7684\u4e0a\u6e38\u4efb\u52a1\u5f80\u5f80\u6bd4\u5c0f DAG \u4e2d\u7684\u4efb\u52a1\u66f4\u53d7\u9752\u7750\u3002\u56e0\u6b64\uff0c\u4f7f\u7528 priority_weight \u9700\u8981\u5bf9\u73af\u5883\u4e2d\u8fd0\u884c\u7684\u5176\u4ed6 DAG \u6709\u4e00\u5b9a\u4e86\u89e3\u3002Celery \u961f\u5217\u548c\u5b64\u7acb\u7684\u5de5\u4f5c\u5668<\/p>\n\n\n\n
    \u5982\u679c\u4f60\u9700\u8981\u4f60\u7684\u4efb\u52a1\u5728\u4e0d\u540c\u7684\u73af\u5883\u4e2d\u6267\u884c\uff08\u4f8b\u5982\uff0c\u4f9d\u8d56\u4e0d\u540c\u7684 python \u5e93\uff0c\u5bc6\u96c6\u578b\u4efb\u52a1\u6709\u66f4\u9ad8\u7684\u8d44\u6e90\u5141\u8bb8\u91cf\uff0c\u6216\u8005\u4e0d\u540c\u7684\u5b58\u53d6\u7ea7\u522b\uff09\uff0c\u4f60\u53ef\u4ee5\u521b\u5efa\u989d\u5916\u7684\u961f\u5217\uff0c\u7531\u4f5c\u4e1a\u7684\u4e00\u4e2a\u5b50\u96c6\u63d0\u4ea4\u4efb\u52a1\u3002\u7136\u540e\uff0c\u5355\u72ec\u7684\u5de5\u4f5c\u96c6\u53ef\u4ee5\u88ab\u914d\u7f6e\u4e3a\u4ece\u5355\u72ec\u7684\u961f\u5217\u4e2d\u63d0\u53d6\u3002\u53ef\u4ee5\u4f7f\u7528\u8fd0\u7b97\u7b26\u4e2d\u7684 queue \u53c2\u6570\u5c06\u4efb\u52a1\u5206\u914d\u5230\u4e00\u4e2a\u5355\u72ec\u7684\u961f\u5217\u3002\u8981\u542f\u52a8\u4e00\u4e2a\u4ece\u4e0d\u540c\u961f\u5217\u8fd0\u884c\u4efb\u52a1\u7684\u5de5\u4f5c\u8005\uff0c\u53ef\u4ee5\u4f7f\u7528\u4ee5\u4e0b\u547d\u4ee4\uff1a<\/p>\n\n\n\n
    <\/p>\n\n\n\n
    bashAirflow celery worker -queues <list of queues><\/code><\/code><\/pre>\n\n\n\n
    \u8fd9\u53ef\u4ee5\u5e2e\u52a9\u786e\u4fdd\u654f\u611f\u6216\u9ad8\u4f18\u5148\u7ea7\u7684\u5de5\u4f5c\u8d1f\u8f7d\u6709\u8db3\u591f\u7684\u8d44\u6e90\uff0c\u56e0\u4e3a\u5b83\u4eec\u4e0d\u4f1a\u4e0e\u5176\u4ed6\u5de5\u4f5c\u8d1f\u8f7d\u7ade\u4e89\u5de5\u4f5c\u8005\u7684\u80fd\u529b\u3002<\/p>\n\n\n\n
    \u6c60\u3001\u4f18\u5148\u6743\u548c\u961f\u5217\u7684\u4efb\u4f55\u7ec4\u5408\u5728\u51cf\u5c11\u8d44\u6e90\u4e89\u7528\u65b9\u9762\u90fd\u662f\u6709\u7528\u7684\u3002\u867d\u7136\u6c60\u5141\u8bb8\u9650\u5236\u5355\u4e2a\u5de5\u4f5c\u8d1f\u8f7d\u5185\u7684\u5e76\u53d1\u6027\uff0c\u4f46 priority_weight \u53ef\u4ee5\u7528\u6765\u4f7f\u5355\u4e2a\u4efb\u52a1\u4ee5\u6bd4\u5176\u4ed6\u4efb\u52a1\u66f4\u4f4e\u7684\u5ef6\u8fdf\u8fd0\u884c\u3002\u5982\u679c\u4f60\u9700\u8981\u66f4\u591a\u7684\u7075\u6d3b\u6027\uff0c\u5de5\u4f5c\u8005\u9694\u79bb\u53ef\u4ee5\u5bf9\u6267\u884c\u4efb\u52a1\u7684\u73af\u5883\u8fdb\u884c\u7ec6\u7c92\u5ea6\u7684\u63a7\u5236\u3002<\/p>\n\n\n\n
    \u91cd\u8981\u7684\u662f\u8981\u8bb0\u4f4f\uff0c\u5e76\u4e0d\u662f\u6240\u6709\u7684\u8d44\u6e90\u90fd\u53ef\u4ee5\u5728 Airflow \u4e2d\u88ab\u4ed4\u7ec6\u5206\u914d\uff1a\u8c03\u5ea6\u5668\u541e\u5410\u91cf\u3001\u6570\u636e\u5e93\u5bb9\u91cf\u548c Kubernetes IP \u7a7a\u95f4\u90fd\u662f\u6709\u9650\u7684\u8d44\u6e90\uff0c\u5982\u679c\u4e0d\u521b\u5efa\u9694\u79bb\u73af\u5883\uff0c\u5c31\u65e0\u6cd5\u5728\u6bcf\u4e2a\u5de5\u4f5c\u8d1f\u8f7d\u7684\u57fa\u7840\u4e0a\u8fdb\u884c\u9650\u5236\u3002\u5c55\u671b<\/p>\n\n\n\n
    \u4ee5\u5982\u6b64\u9ad8\u7684\u541e\u5410\u91cf\u8fd0\u884c Airflow\uff0c\u9700\u8981\u8003\u8651\u5f88\u591a\u56e0\u7d20\uff0c\u4efb\u4f55\u89e3\u51b3\u65b9\u6848\u7684\u7ec4\u5408\u90fd\u662f\u6709\u7528\u7684\u3002\u6211\u4eec\u5df2\u7ecf\u5b66\u5230\u4e86\u5f88\u591a\uff0c\u6211\u4eec\u5e0c\u671b\u4f60\u80fd\u8bb0\u4f4f\u8fd9\u4e9b\u6559\u8bad\uff0c\u5e76\u5728\u4f60\u81ea\u5df1\u7684 Airflow \u57fa\u7840\u8bbe\u65bd\u548c\u5de5\u5177\u4e2d\u5e94\u7528\u6211\u4eec\u7684\u4e00\u4e9b\u89e3\u51b3\u65b9\u6848\u3002<\/p>\n\n\n\n
    \u603b\u7ed3\u4e00\u4e0b\u6211\u4eec\u7684\u4e3b\u8981\u6536\u83b7\uff1a<\/p>\n\n\n\n