From fdae074dc616c03ebfb37591f9b9526d2dcb2539 Mon Sep 17 00:00:00 2001 From: hcl01387923 Date: Tue, 21 Jul 2026 15:08:48 +0800 Subject: [PATCH 1/2] fix --- aliyun/log/scheduled_sql.py | 16 ++++++++++++- .../schedule_sql/schedule_sql_demo_test.py | 23 +++++++++++-------- 2 files changed, 28 insertions(+), 11 deletions(-) diff --git a/aliyun/log/scheduled_sql.py b/aliyun/log/scheduled_sql.py index 0e981914..ce084295 100644 --- a/aliyun/log/scheduled_sql.py +++ b/aliyun/log/scheduled_sql.py @@ -84,10 +84,24 @@ def configurationToDict(self): "toTime": self.__toTime, "dataFormat": self.__dataFormat, "parameters": self.__parameters if isinstance(self.__parameters, dict) else self.__parameters.toDict( - self.__dataFormat) + self.__dataFormat), + "forceComplete": self.__forceComplete, + "usingExactlyOnce": self.__usingExactlyOnce } return config + def getForceComplete(self): + return self.__forceComplete + + def setForceComplete(self, forceComplete): + self.__forceComplete = forceComplete + + def getUsingExactlyOnce(self): + return self.__usingExactlyOnce + + def setUsingExactlyOnce(self, usingExactlyOnce): + self.__usingExactlyOnce = usingExactlyOnce + def getSourceLogstore(self): return self.__sourceLogstore diff --git a/tests/e2e/schedule_sql/schedule_sql_demo_test.py b/tests/e2e/schedule_sql/schedule_sql_demo_test.py index 74579dcb..38b4faca 100644 --- a/tests/e2e/schedule_sql/schedule_sql_demo_test.py +++ b/tests/e2e/schedule_sql/schedule_sql_demo_test.py @@ -9,17 +9,17 @@ accessKeyId = "" # The AccessKeyId accessKeySecret = "" # The AccessKeySecret -endpoint = "cn-shanghai.log.aliyuncs.com" # The source endpoint of the project's region -roleArn = "" # The roleArn -project = "etl-project" # The source project name -source_logstore = "" # The source logstore name +endpoint = "cn-heyuan.log.aliyuncs.com" # The source endpoint of the project's region +roleArn = "acs:ram::111:role/aliyunlogetlrole" # The roleArn +project = "111" # The source project name +source_logstore = "111" # The source logstore name source_metricstore = "" # The source metricstore name dest_endpoint = "" # The endpoint of the destination project's region -dest_role_arn = "" # The destination roleArn -dest_project = "" # The destination project name +dest_role_arn = "acs:ram::111:role/aliyunlogetlrole" # The destination roleArn +dest_project = "111" # The destination project name dest_logstore = "" # The destination logstore name -dest_metricstore = "" # The destination metricstore name +dest_metricstore = "111" # The destination metricstore name from_time = int(time.time()) - 360 # The start time of the scheduled SQL task @@ -34,7 +34,7 @@ delay_seconds = 0 # the delay seconds for schedule sql # three possible values for the variable data_format : "log2log" , "log2metric" ,"metric2metric" -data_format = "log2log" +data_format = "log2metric" # Possible values for the variable schedule_type: "FixedRate", "Daily", "Weekly", "Hourly", "Cron" # schedule_type = "FixedRate" @@ -79,6 +79,8 @@ def generate_schedule_sql(): config.setToTime(0) config.setDataFormat(data_format) config.setParameters(parameters) + config.setForceComplete(True) + config.setUsingExactlyOnce(True) schedule_sql.setSchedule(schedule_rule) schedule_sql.setConfiguration(config) @@ -314,6 +316,7 @@ def update_schedule_sql(): configuration['fromTimeExpr'] = '@m-3m' configuration['maxRetries'] = 25 + configuration['usingExactlyOnce'] = True schedule['type'] = 'Cron' schedule['timeZone'] = '+0800' @@ -331,11 +334,11 @@ def update_schedule_sql(): schedule_sql.setName(job_name) schedule_sql.setDisplayName(display_name) schedule_sql.setDescription(description) - + print(configuration.configurationToDict()) update_schedule_sql_response = client.update_scheduled_sql(project, schedule_sql) update_schedule_sql_response.log_print() if __name__ == "__main__": - create_schedule_sql() + update_schedule_sql() pass From 8e738a036e40bbc7b0fda45ddc0e59183a9d3890 Mon Sep 17 00:00:00 2001 From: hcl01387923 Date: Mon, 27 Jul 2026 10:18:50 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=8F=82=E6=95=B0?= =?UTF-8?q?=E5=88=9D=E5=A7=8B=E5=8C=96=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- aliyun/log/scheduled_sql.py | 2 ++ tests/e2e/schedule_sql/schedule_sql_demo_test.py | 13 ++++++------- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/aliyun/log/scheduled_sql.py b/aliyun/log/scheduled_sql.py index ce084295..5dcd0e99 100644 --- a/aliyun/log/scheduled_sql.py +++ b/aliyun/log/scheduled_sql.py @@ -63,6 +63,8 @@ def __init__(self): self.__fromTime = 0 self.__toTime = 0 self.__dataFormat = "log2log" + self.__forceComplete = False + self.__usingExactlyOnce = False self.__parameters = ScheduledSQLParameters() def configurationToDict(self): diff --git a/tests/e2e/schedule_sql/schedule_sql_demo_test.py b/tests/e2e/schedule_sql/schedule_sql_demo_test.py index 38b4faca..413534e8 100644 --- a/tests/e2e/schedule_sql/schedule_sql_demo_test.py +++ b/tests/e2e/schedule_sql/schedule_sql_demo_test.py @@ -10,16 +10,16 @@ accessKeySecret = "" # The AccessKeySecret endpoint = "cn-heyuan.log.aliyuncs.com" # The source endpoint of the project's region -roleArn = "acs:ram::111:role/aliyunlogetlrole" # The roleArn -project = "111" # The source project name -source_logstore = "111" # The source logstore name +roleArn = "" # The roleArn +project = "etl-project" # The source project name +source_logstore = "" # The source logstore name source_metricstore = "" # The source metricstore name dest_endpoint = "" # The endpoint of the destination project's region -dest_role_arn = "acs:ram::111:role/aliyunlogetlrole" # The destination roleArn -dest_project = "111" # The destination project name +dest_role_arn = "" # The destination roleArn +dest_project = "" # The destination project name dest_logstore = "" # The destination logstore name -dest_metricstore = "111" # The destination metricstore name +dest_metricstore = "" # The destination metricstore name from_time = int(time.time()) - 360 # The start time of the scheduled SQL task @@ -334,7 +334,6 @@ def update_schedule_sql(): schedule_sql.setName(job_name) schedule_sql.setDisplayName(display_name) schedule_sql.setDescription(description) - print(configuration.configurationToDict()) update_schedule_sql_response = client.update_scheduled_sql(project, schedule_sql) update_schedule_sql_response.log_print()