dags/test_datax_task.py

from airflow import DAG
from datetime import datetime
from airflow.operators.bash import BashOperator
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.configuration import conf

namespace = conf.get("kubernetes", "NAMESPACE")


# set the name that will be printed
name = "luoyulong"

# instantiate the DAG
with DAG(
        start_date=datetime(2022,6,1),
        catchup=False,
        schedule_interval='@daily',
        dag_id="test_auto_dag_datax"
) as dag:
    
        op_xxx_yyy_zzz = KubernetesPodOperator(
        task_id="xxx_yyy_zzz",
        image="SXKJ:32775/pod_datax:0.9",
        in_cluster=True,
        namespace=namespace,
        name="data_sync",
        random_name_suffix=True,
        labels={'app':'backend', 'env':'dev'},
        reattach_on_restart=True,
        is_delete_operator_pod=False,
        get_logs=True,
        log_events_on_failure=True,
        cmds=['/bin/bash', '-c', 'cd datax/bin && echo $SCRIPT > config.json && echo "\'"$HOME/conda/envs/py27/bin/python datax.py $CMD_PARM config.json"\'" |xargs bash -c  '],
        env_vars={'SCRIPT': '{\n  "job": {\n    "setting": {\n      "speed": {\n        "channel": 3\n      },\n      "errorLimit": {\n        "record": 0,\n        "percentage": 0.02\n      }\n    },\n    "content": [\n      {\n        "reader": {\n          "name": "mysqlreader",\n          "parameter": {\n            "username": "root",\n            "password": "happylay",\n            "column": [\n              "id",\n              "name"\n            ],\n            "where": "ct=\'shanghai\'",\n            "splitPk": "",\n            "connection": [\n              {\n                "table": [\n                  "test_3"\n                ],\n                "jdbcUrl": [\n                  "jdbc:mysql://192.168.199.107:10086/test-db?useSSL=false"\n                ]\n              }\n            ]\n          }\n        },\n        "writer": {\n          "name": "hdfswriter",\n          "parameter": {\n            "defaultFS": "hdfs://192.168.199.27:9000",\n            "fileType": "text",\n            "path": "/home/sxkj/bigdata/apache-hive-2.3.9-bin/warehouse/test_p/ct=shanghai",\n            "fileName": "test_p",\n            "writeMode": "append",\n            "fieldDelimiter": ",",\n            "column": [\n              {\n                "name": "id",\n                "type": "int"\n              },\n              {\n                "name": "name",\n                "type": "string"\n              }\n            ]\n          }\n        }\n      }\n    ]\n  }\n}', 'CMD_PARM': '-j "-Xms2G -Xmx2G"  -p "-Dct=hangzhou" '}
        )