创建定时SQL任务

def create_scheduled_sql(self, project_name, scheduled_sql):

import os
import time
from aliyun.log import LogClient
from aliyun.log.scheduled_sql import *

accessKeyId = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')  # The AccessKeyId
accessKeySecret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')  # The AccessKeySecret

endpoint = "cn-hangzhou.log.aliyuncs.com"  # The  source endpoint of the project's region
roleArn = "acs:ram::141******5616316:role/aliyunserviceroleforslsaudit"  # The roleArn
project = "demo-test-project"  # The source project name
source_logstore = "test-logstore"  # The source logstore name
source_metricstore = ""  # The source metricstore name

dest_endpoint = "cn-hangzhou.log.aliyuncs.com"  # The endpoint of the destination project's region
dest_role_arn = "acs:ram::141******5616316:role/aliyunserviceroleforslsaudit"  # The destination roleArn
dest_project = "demo-test-project"  # The destination project name
dest_logstore = "test-logstore2"  # The destination logstore name
dest_metricstore = ""  # The destination metricstore name

from_time = int(time.time()) - 360  # The start time of the scheduled SQL task

job_name = "test-001"  # The job name
display_name = "test-001"  # The display name
description = "创建一个定时SQL任务"  # The description

script = "* | select *"  # The SQL script
promql = "* | select promql_query_range('key{}') from metrics limit 1000"  # The PromQL script

instance_id = ""  # The job instanceId for schedule sql
delay_seconds = 0  # the delay seconds for schedule sql

# three possible values for the variable data_format  : "log2log" , "log2metric" ,"metric2metric"
data_format = "log2log"

# Possible values for the variable schedule_type: "FixedRate", "Daily", "Weekly", "Hourly", "Cron"
# schedule_type = "FixedRate"
schedule_type = "FixedRate"

client = LogClient(endpoint, accessKeyId, accessKeySecret)


def generate_schedule_sql():
    schedule_rule = generate_schedule_rule()
    config = generate_config()

    schedule_sql = ScheduledSQL()
    schedule_sql.setName(job_name)
    schedule_sql.setDisplayName(display_name)
    schedule_sql.setDescription(description)
    schedule_sql.setSchedule(schedule_rule)
    schedule_sql.setConfiguration(config)
    return schedule_sql


def generate_schedule_rule():
    schedule_rule = JobSchedule()
    schedule_rule.setType(schedule_type)
    schedule_rule.setDelay(delay_seconds)
    schedule_rule.setTimeZone('+0800')
    if schedule_type == "FixedRate":
        # The interval value must be a string in the format "Xs", "Xm", "Xh", or "Xd",
        # where X is an integer representing the number of seconds, minutes, hours, or days.
        schedule_rule.setInterval('60s')
    elif schedule_type == "Cron":
        # The cron expression must be a valid cron string.
        schedule_rule.setCronExpression("*/2 * * * *")
    elif schedule_type == "Daily":
        # The hour value must be an integer between 0 and 23.
        schedule_rule.setHour(1)
    elif schedule_type == "Weekly":
        # The day of the week value must be an integer between 1 and 7.
        # The hour value must be an integer between 0 and 23.
        schedule_rule.setDayOfWeek(1)
        schedule_rule.setHour(2)
    elif schedule_type == "Hourly":
        pass

    return schedule_rule


def generate_config():
    query = script
    if data_format == 'log2log':
        source = source_logstore
        dest = dest_logstore
    elif data_format == 'log2metric':
        source = source_logstore
        dest = dest_metricstore
    elif data_format == 'metric2metric':
        source = source_metricstore
        dest = dest_metricstore
        query = promql

    parameters = generate_log2log_params()
    config = ScheduledSQLConfiguration()
    config.setScript(query)
    config.setSqlType('searchQuery')
    config.setDestEndpoint(dest_endpoint)
    config.setDestProject(dest_project)
    config.setSourceLogstore(source)
    config.setDestLogstore(dest)
    config.setRoleArn(roleArn)
    config.setDestRoleArn(dest_role_arn)
    config.setFromTimeExpr("@m - 5m")
    config.setToTimeExpr("@m")
    config.setMaxRunTimeInSeconds(1800)
    config.setResourcePool("enhanced")
    config.setMaxRetries(10)
    config.setFromTime(from_time)
    config.setToTime(0)
    config.setDataFormat(data_format)
    config.setParameters(parameters)
    return config


def generate_log2log_params():
    parameters = ScheduledSQLBaseParameters()
    return parameters


if __name__ == '__main__':
    schedule_sql = generate_schedule_sql()
    res = client.create_scheduled_sql(dest_project, schedule_sql)
    res.log_print()

CreateScheduleSqlResponse:
headers: {'Server': 'AliyunSLS', 'Content-Length': '0', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Fri, 01 Nov 2024 07:04:39 GMT', 'x-log-time': '1730444679', 'x-log-requestid': '67247D87DB04A454D92A6F6B'}

Process finished with exit code 0