batch_test.py 1.14 KB
Newer Older
김진영's avatar
김진영 committed
1
2
from datetime import datetime, timedelta
from textwrap import dedent
김진영's avatar
김진영 committed
3
import pendulum
김진영's avatar
김진영 committed
4
from airflow.operators.bash import BashOperator
김진영's avatar
김진영 committed
5
from airflow import DAG
김진영's avatar
김진영 committed
6
from data.gasan_data import gasanData   # import cj-gasan data
김진영's avatar
김진영 committed
7

김진영's avatar
김진영 committed
8
# set timezone
김진영's avatar
김진영 committed
9
10
local_tz = pendulum.timezone("Asia/Seoul")

김진영's avatar
김진영 committed
11
12
13
14
with DAG(
    'batch_test',
    default_args={
        'depends_on_past': False,
김진영's avatar
김진영 committed
15
        'email': 'kim-jy@lotte.net',
김진영's avatar
김진영 committed
16
17
    },
    description='Test Batch Job',
김진영's avatar
김진영 committed
18
    schedule_interval='*/1 * * * *',
김진영's avatar
김진영 committed
19
    start_date=datetime(2022, 5, 13, tzinfo=local_tz),
김진영's avatar
김진영 committed
20
    tags=['test'],
김진영's avatar
김진영 committed
21
    catchup=False,
김진영's avatar
김진영 committed
22
) as dag:
김진영's avatar
김진영 committed
23
24
    # gasan 작업 병렬처리
    post_gasan_tasks = []
김진영's avatar
김진영 committed
25

김진영's avatar
김진영 committed
26
    for i, data in enumerate(gasanData):
김진영's avatar
김진영 committed
27
        post_gasan_task = BashOperator(
김진영's avatar
김진영 committed
28
29
            task_id='post_gasan'+str(i+1),
            bash_command="curl -X \'POST\' \'http://10.231.238.224:30999/api/v1/camera/writeimage\' -H \'Content-Type: application/json\' -d \'{\"id\": \"test\", \"pw\": \"test\", \"ip\": \"%s\", \"serialNum\": \"%s\", \"camName\": \"%s\"}\'" %(data["ip"], data["serialNum"], data["camName"]),
김진영's avatar
김진영 committed
30
        )
김진영's avatar
김진영 committed
31

김진영's avatar
김진영 committed
32
        post_gasan_tasks.append(post_gasan_task)
김진영's avatar
김진영 committed
33

김진영's avatar
김진영 committed
34
    # 작업 순서 정의
김진영's avatar
김진영 committed
35
    post_gasan_tasks