You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "Zkplo (via GitHub)" <gi...@apache.org> on 2024/11/30 02:46:01 UTC
[PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Zkplo opened a new pull request, #11568:
URL: https://github.com/apache/inlong/pull/11568
<!-- Prepare a Pull Request
Change the title of pull request refer to the following example:
[INLONG-XYZ][Component] Title of the pull request
-->
<!-- Specify the issue this pull request going to fix.
The following *XYZ* should be replaced by the actual [GitHub Issue](https://github.com/apache/inlong/issues) number)-->
Fixes #11567
### Motivation
Optimize the original DAGs of Airflow.
### Modifications
<!--Describe the modifications you've done.-->
### Verifying this change
*(Please pick either of the following options)*
- [x] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [ ] This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1865276893
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,47 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+
+ conf = context.get('dag_run').conf
+ groupId = conf.get('inlong_group_id')
+
+ logging.info(f"Execution parameters for this cleaning task: {conf}")
+ logging.info(f"The execution time of this cleaning task is: {current_time} for groupId = {groupId}")
Review Comment:
```suggestion
logging.info(f"Execution parameters = {conf} for groupId = {groupId} and execution time = {current_time}")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "Zkplo (via GitHub)" <gi...@apache.org>.
Zkplo commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1865331335
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,47 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+
+ conf = context.get('dag_run').conf
+ groupId = conf.get('inlong_group_id')
+
+ logging.info(f"Execution parameters for this cleaning task: {conf}")
+ logging.info(f"The execution time of this cleaning task is: {current_time} for groupId = {groupId}")
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "Zkplo (via GitHub)" <gi...@apache.org>.
Zkplo commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1865161288
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+ logging.info(f"The execution time of this cleaning task is: {current_time}")
Review Comment:
done.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+ logging.info(f"The execution time of this cleaning task is: {current_time}")
+
+ conf = context.get('dag_run').conf
+ logging.info(f"Execution parameters for this cleaning task: {conf}")
+ groupId = conf.get('inlong_group_id')
+
+ if groupId is None or len(groupId) == 0:
+ for dag_file in os.listdir(DAG_PATH):
+ if dag_file.endswith(".py") and dag_file.startswith(DAG_PREFIX):
+ dag_file_path = os.path.join(DAG_PATH, dag_file)
+ with open(dag_file_path, "r") as file:
line = file.readline()
- end_date_str = None
- if len(line.split("=")) > 1:
- end_date_str = line.split("=")[1].strip().strip("\"")
- logging.info(f"DAG end time: {end_date_str}")
- if end_date_str:
+ while line and "end_offset_datetime_str" not in line:
+ line = file.readline()
+ end_date_str = None
+ row = line.split("=")
+ if len(row) > 1:
+ end_date_str = datetime.fromtimestamp(int(row[1].strip().strip("\"")) / 1000, tz=target_timezone)
+ logging.info(f"The end time of '{dag_file}' is: {end_date_str}")
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1864904186
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_creator.py:
##########
@@ -98,7 +96,7 @@ def taskFunction(**context):
print("Request Body: ", data)
response = requests.post(url, params=params, headers=headers, json=data)
if response.status_code == 200:
- print(response.json())
+ print(response.json()) # Assuming to return JSON data
Review Comment:
Please print more detailed information.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
logging.info(f"Current time: {current_time}")
Review Comment:
Ditto.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+
+ conf = context.get('dag_run').conf
+ logging.info(f"conf: {conf}")
+ groupId = conf.get('inlong_group_id')
+
+ if groupId is None or len(groupId) == 0:
+ for dag_file in os.listdir(DAG_PATH):
+ if dag_file.endswith(".py") and dag_file.startswith(DAG_PREFIX):
+ dag_file_path = os.path.join(DAG_PATH, dag_file)
+ with open(dag_file_path, "r") as file:
line = file.readline()
- end_date_str = None
- if len(line.split("=")) > 1:
- end_date_str = line.split("=")[1].strip().strip("\"")
- logging.info(f"DAG end time: {end_date_str}")
- if end_date_str:
+ while line and "end_offset_datetime_str" not in line:
+ line = file.readline()
+ end_date_str = None
+ row = line.split("=")
+ if len(row) > 1:
+ end_date_str = datetime.fromtimestamp(int(row[1].strip().strip("\"")) / 1000, tz=target_timezone)
+ logging.info(f"DAG end time: {end_date_str}")
Review Comment:
Ditto.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+
+ conf = context.get('dag_run').conf
+ logging.info(f"conf: {conf}")
Review Comment:
Ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang merged PR #11568:
URL: https://github.com/apache/inlong/pull/11568
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "Zkplo (via GitHub)" <gi...@apache.org>.
Zkplo commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1865122777
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_creator.py:
##########
@@ -98,7 +96,7 @@ def taskFunction(**context):
print("Request Body: ", data)
response = requests.post(url, params=params, headers=headers, json=data)
if response.status_code == 200:
- print(response.json())
+ print(response.json()) # Assuming to return JSON data
Review Comment:
sure. done.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+
+ conf = context.get('dag_run').conf
+ logging.info(f"conf: {conf}")
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [INLONG-11567][Manager] Optimize the original DAG of Airflow [inlong]
Posted by "fuweng11 (via GitHub)" <gi...@apache.org>.
fuweng11 commented on code in PR #11568:
URL: https://github.com/apache/inlong/pull/11568#discussion_r1865131263
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+ logging.info(f"The execution time of this cleaning task is: {current_time}")
+
+ conf = context.get('dag_run').conf
+ logging.info(f"Execution parameters for this cleaning task: {conf}")
+ groupId = conf.get('inlong_group_id')
+
+ if groupId is None or len(groupId) == 0:
+ for dag_file in os.listdir(DAG_PATH):
+ if dag_file.endswith(".py") and dag_file.startswith(DAG_PREFIX):
+ dag_file_path = os.path.join(DAG_PATH, dag_file)
+ with open(dag_file_path, "r") as file:
line = file.readline()
- end_date_str = None
- if len(line.split("=")) > 1:
- end_date_str = line.split("=")[1].strip().strip("\"")
- logging.info(f"DAG end time: {end_date_str}")
- if end_date_str:
+ while line and "end_offset_datetime_str" not in line:
+ line = file.readline()
+ end_date_str = None
+ row = line.split("=")
+ if len(row) > 1:
+ end_date_str = datetime.fromtimestamp(int(row[1].strip().strip("\"")) / 1000, tz=target_timezone)
+ logging.info(f"The end time of '{dag_file}' is: {end_date_str}")
Review Comment:
```suggestion
logging.info(f"The end time of '{dag_file}' is: {end_date_str} for groupId={groupId}")
```
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+ logging.info(f"The execution time of this cleaning task is: {current_time}")
+
+ conf = context.get('dag_run').conf
+ logging.info(f"Execution parameters for this cleaning task: {conf}")
+ groupId = conf.get('inlong_group_id')
+
+ if groupId is None or len(groupId) == 0:
+ for dag_file in os.listdir(DAG_PATH):
+ if dag_file.endswith(".py") and dag_file.startswith(DAG_PREFIX):
+ dag_file_path = os.path.join(DAG_PATH, dag_file)
+ with open(dag_file_path, "r") as file:
line = file.readline()
- end_date_str = None
- if len(line.split("=")) > 1:
- end_date_str = line.split("=")[1].strip().strip("\"")
- logging.info(f"DAG end time: {end_date_str}")
- if end_date_str:
+ while line and "end_offset_datetime_str" not in line:
+ line = file.readline()
+ end_date_str = None
+ row = line.split("=")
+ if len(row) > 1:
+ end_date_str = datetime.fromtimestamp(int(row[1].strip().strip("\"")) / 1000, tz=target_timezone)
+ logging.info(f"The end time of '{dag_file}' is: {end_date_str}")
try:
- if str(current_time) > str(end_date_str):
- dag_file_path = os.path.join(DAG_PATH, dag_file)
+ if end_date_str and str(current_time) > str(end_date_str):
os.remove(dag_file_path)
- # Optionally, delete the end_date variable
logging.info(f"Deleted expired DAG: {dag_file}")
except ValueError:
- logging.error(f"Invalid date format for DAG {dag_file}: {end_date_str}")
+ logging.error(f"Failed to delete {dag_file}: {end_date_str}")
Review Comment:
Ditto.
##########
inlong-manager/manager-schedule/src/test/resources/airflow/dag_cleaner.py:
##########
@@ -29,33 +29,46 @@
from airflow import configuration
DAG_PATH = configuration.get('core', 'dags_folder') + "/"
-
+DAG_PREFIX = 'inlong_offline_task_'
def clean_expired_dags(**context):
+
original_time = context.get('execution_date')
target_timezone = pytz.timezone("Asia/Shanghai")
utc_time = original_time.astimezone(target_timezone)
- current_time = utc_time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
- logging.info(f"Current time: {current_time}")
- for dag_file in os.listdir(DAG_PATH):
- if dag_file.endswith(".py") and dag_file.startswith("inlong_offline_task_"):
- with open(DAG_PATH + dag_file, "r") as file:
- line = file.readline()
- while line and "end_offset_datetime_str" not in line:
+ current_time = utc_time.strftime("%Y-%m-%d %H:%M:%S.%f")
+ logging.info(f"The execution time of this cleaning task is: {current_time}")
Review Comment:
```suggestion
logging.info(f"The execution time of this cleaning task is: {current_time} for groupId={groupId}")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org